[FLINK-40354][state/forst] Configure the TTL compaction filter for State V2 states - #29149
Open
seungjoo-choi-bucketplace wants to merge 1 commit into
Conversation
…ate V2 states On the State V2 (async) path of the ForSt state backend the TTL compaction filter factory is attached to the column family when a TTL state is registered (ForStDBTtlCompactFiltersManager#setAndRegisterCompactFilterIfStateTtlV2), but the filter never receives its configuration: configCompactFilter() only accepts the V1 StateDescriptor and is only called by the sync backends. An unconfigured FlinkCompactionFilter stays disabled and keeps every entry, so expired state is never physically removed and state size grows without bound despite StateTtlConfig. This adds a configCompactFilter overload for the V2 StateDescriptor (sharing the existing body) and calls it from ForStKeyedStateBackend#createStateInternal right after the state is registered, mirroring ForStSyncKeyedStateBackend and RocksDBKeyedStateBackend. A compactState() test hook is added to the async backend as well, and ForStTtlCompactFilterTest verifies that expired entries of a V2 ValueState are dropped by compaction. Co-Authored-By: Claude Code <noreply@anthropic.com>
seungjoo-choi-bucketplace
marked this pull request as ready for review
September 9, 2026 23:59
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Fixes FLINK-40354: with the ForSt state backend,
StateTtlConfigbackground 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
FlinkCompactionFiltertakes two steps — (1) attach aFlinkCompactionFilterFactoryto the column family, (2) push the TTL configuration (ttl, state type,query-time-after-num-entries) into that factory viaForStDBTtlCompactFiltersManager#configCompactFilter. On the V2 path only step (1) happens (setAndRegisterCompactFilterIfStateTtlV2inForStOperationUtils#createColumnFamilyDescriptor).configCompactFilteronly accepts the V1org.apache.flink.api.common.state.StateDescriptorand is only called byForStSyncKeyedStateBackend/RocksDBKeyedStateBackend;ForStKeyedStateBackend#createStateInternalnever calls it. An unconfiguredFlinkCompactionFilterstays disabled and keeps every entry. This affects every TTL state type on the V2 path (Value/List/Map), not onlyMapStateas 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 aconfigCompactFilteroverload taking the V2org.apache.flink.api.common.state.v2.StateDescriptor; both overloads delegate to a shared private implementation (state type is derived fromStateDescriptor.Typefor V2, from the descriptor class for V1 as before).ForStKeyedStateBackend#createStateInternal: callconfigCompactFilterright aftertryRegisterKvStateInformation, mirroring the sync backends. Add a@VisibleForTesting compactState(StateDescriptor)hook (same as the sync backend has).ForStTestUtils:createKeyedStateBackendoverload that accepts aTtlTimeProvider.ForStTtlCompactFilterTest.Verifying this change
This change added tests and can be verified as follows:
ForStTtlCompactFilterTest#testExpiredEntriesAreRemovedByCompaction: creates a V2ValueStatewith TTL (cleanupInRocksdbCompactFilter,ReturnExpiredIfNotCleanedUpso the read path does not mask expired entries), writes entries, advances a controllableTtlTimeProviderpast the TTL, writes a fresh entry, triggerscompactState()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.ValueStatewith 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:
@Public(Evolving): noDocumentation
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.