Hello, I have a question about this code snipit and text from the Flink docs: 1. If I am using keyed state, does

cleanupInRocksdbCompactFilter

trigger when 1000 state entries for a given key are processed? Or 1000 across all keys globally in the application? 2. What exactly is meant by "processing a state entry"? I am assuming it means when a record is added to state 3. Could I specify both

cleanupInRocksdbCompactFilter(..)

and

cleanupFullSnapshot

? (assuming yes based on docs but wanted to confirm) https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/#incremental-cleanup

import org.apache.flink.api.common.state.StateTtlConfig;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .cleanupInRocksdbCompactFilter(1000)
    .build();

RocksDB compaction filter will query current timestamp, used to check expiration, from Flink every time after processing certain number of state entries. You can change it and pass a custom value to StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries) method. Updating the timestamp more often can improve cleanup speed but it decreases compaction performance because it uses JNI call from native code. The default background cleanup for RocksDB backend queries the current timestamp each time 1000 entries have been processed.