[FLINK-40354][state/forst] Configure the TTL compaction filter for State V2 states

[seungjoo-choi-bucketplace]

What is the purpose of the change

Fixes FLINK-40354: with the ForSt state backend, StateTtlConfig background cleanup does not work for states created through the State V2 (async) API — expired entries are never physically removed and the state grows without bound.

Root cause: enabling the native FlinkCompactionFilter takes two steps — (1) attach a FlinkCompactionFilterFactory to the column family, (2) push the TTL configuration (ttl, state type, query-time-after-num-entries) into that factory via ForStDBTtlCompactFiltersManager#configCompactFilter. On the V2 path only step (1) happens (setAndRegisterCompactFilterIfStateTtlV2 in ForStOperationUtils#createColumnFamilyDescriptor). configCompactFilter only accepts the V1 org.apache.flink.api.common.state.StateDescriptor and is only called by ForStSyncKeyedStateBackend / RocksDBKeyedStateBackend; ForStKeyedStateBackend#createStateInternal never calls it. An unconfigured FlinkCompactionFilter stays disabled and keeps every entry. This affects every TTL state type on the V2 path (Value/List/Map), not only MapState as reported in the ticket. The code is the same on release-2.1, release-2.2, release-2.3 and master.

Brief change log

• ForStDBTtlCompactFiltersManager: add a configCompactFilter overload taking the V2 org.apache.flink.api.common.state.v2.StateDescriptor; both overloads delegate to a shared private implementation (state type is derived from StateDescriptor.Type for V2, from the descriptor class for V1 as before).
• ForStKeyedStateBackend#createStateInternal: call configCompactFilter right after tryRegisterKvStateInformation, mirroring the sync backends. Add a @VisibleForTesting compactState(StateDescriptor) hook (same as the sync backend has).
• ForStTestUtils: createKeyedStateBackend overload that accepts a TtlTimeProvider.
• New ForStTtlCompactFilterTest.

Verifying this change

This change added tests and can be verified as follows:

• ForStTtlCompactFilterTest#testExpiredEntriesAreRemovedByCompaction: creates a V2 ValueState with TTL (cleanupInRocksdbCompactFilter, ReturnExpiredIfNotCleanedUp so the read path does not mask expired entries), writes entries, advances a controllable TtlTimeProvider past the TTL, writes a fresh entry, triggers compactState() and asserts that only the expired entries are gone. The test fails on master without the fix (expired entry k1 should be removed) and passes with it.
• Production observation (Flink 2.1.3, ForSt on S3, State V2 ValueState with a 1-day TTL, ~45 GB/day ingestion of write-once keys): before the fix the checkpoint size grew linearly at the ingestion rate for 29 hours although compaction was running (write amplification ~14x); with the fix applied through a class override, growth stopped within an hour and turned negative as bottom-level files were rewritten. TaskManagers that were still on the unpatched image kept growing at the ingestion rate, which gave an A/B confirmation.

No state format change is involved — the TTL timestamp is already part of the serialized value — so existing state is cleaned up on the next compactions after upgrading.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — state TTL cleanup now actually happens for ForSt V2 states (compaction filter active)
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Note: my ASF JIRA account is still pending approval, so I could not yet ask for the ticket to be assigned — opening this as a draft in the meantime. The analysis and the patch were prepared with AI assistance (Claude Code) and validated in production as described above.

[flinkbot]

CI report:

• 1c897c3 UNKNOWN
• 0efcc76 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[minxhe]

V2 ListState passes a TtlSerializer<E> directly, not a ListSerializer<TtlValue<E>>, so the new V2 call reaches an invalid cast:

((ListSerializer<?>) stateSerializer) // stateSerializer is TtlSerializer<E>

This introduces a ClassCastException during TTL-enabled V2 ListState creation. Could we use the serializer directly for V2, unwrap it only for V1, and add a V2 ListState regression test?

[seungjoo-choi-bucketplace]

Good catch, thanks — you're right: the V2 ListStateDescriptor carries the element serializer itself (TtlSerializer<E> for a TTL state), so the unwrap only applies to V1.

Fixed in 1c897c3: the V1 overload unwraps the ListSerializer, the V2 overload passes the registered serializer through as the element serializer (which matches how ForStListState stores the elements via ListDelimitedSerializer), and ForStTtlCompactFilterTest now covers a V2 ListState for both variable-length (element filter) and fixed-length elements. Both new tests fail on the previous revision with exactly this ClassCastException.