seungjoo-choi-bucketplace opened a new pull request, #29149: URL: https://github.com/apache/flink/pull/29149
## What is the purpose of the change Fixes [FLINK-40354](https://issues.apache.org/jira/browse/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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
