import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.contrib.streaming.state.RocksDBConfigSetter;

public class CustomRocksDBConfigSetter implements RocksDBConfigSetter {

    @Override
    public void setConfig(String propertyName, String propertyValue) {
        // You can set other RocksDB options here if needed
    }

    @Override
    public void setOptions(DBOptions options) {
        // Set a custom compaction filter factory
        options.setCompactionFilterFactory(new NativeCompactionFilterFactory() {
            @Override
            public CompactionFilter createCompactionFilter(String columnFamilyName) {
                return new YourCustomCompactionFilter(); // Replace with your custom filter class
            }
        });
    }

    @Override
    public void setColumnFamilyOptions(ColumnFamilyOptions options, String columnFamilyName) {
        // Additional configurations for specific column families can be set here
    }
}

import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;

// Inside your Flink job setup
RocksDBStateBackend rocksDBBackend = new RocksDBStateBackend(
    new Path("/path/to/checkpoint_directory"), // Specify your checkpoint directory
    true // Enable incremental checkpoints
);

// Apply the custom RocksDB configuration
rocksDBBackend.setRocksDBConfigSetter(new CustomRocksDBConfigSetter());

StateTtlConfig#enableCleanupInBackground(true)

state.backend.rocksdb.total-delayed-state-size

num-delayed-keys

write-buffer-size, and max-write-buffer-number