[jira] [Updated] (FLINK-41093) AsyncKeyedStateBackendAdaptor ignores the TTL config of State V2 descriptors

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-41093?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ASF GitHub Bot updated FLINK-41093:
-----------------------------------
    Labels: pull-request-available  (was: )

> AsyncKeyedStateBackendAdaptor ignores the TTL config of State V2 descriptors
> ----------------------------------------------------------------------------
>
>                 Key: FLINK-41093
>                 URL: https://issues.apache.org/jira/browse/FLINK-41093
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / State Backends
>            Reporter: Aaron He
>            Priority: Major
>              Labels: pull-request-available
>
> h2. Problem
> When the State V2 (async state) API runs on a state backend without native async state support, that is HashMap or RocksDB, Flink wraps the keyed state backend in AsyncKeyedStateBackendAdaptor. Keyed state created through this adaptor ignores the TTL configured with StateDescriptor#enableTimeToLive. Values never expire and are never cleaned up.
> SQL operators with table.exec.async-state.enabled set a state TTL this way, for example AsyncStateGroupAggFunction. On HashMap or RocksDB they keep state beyond the configured retention, so state can grow without bound, and their results differ from the sync operators once the TTL has passed. ForSt implements State V2 natively and is not affected.
> h2. Cause
> AsyncKeyedStateBackendAdaptor#createStateInternal converts the V2 descriptor with StateDescriptorUtils#transformFromV2ToV1 and passes the result to KeyedStateBackend#getPartitionedState. The conversion copies only the state name and the serializer or type information, so the V1 descriptor has TTL disabled and TtlStateFactory never wraps the state. This holds for value, list, map, reducing and aggregating state.
> h2. Proposed fix
> Copy the V2 descriptor's StateTtlConfig onto the converted V1 descriptor in AsyncKeyedStateBackendAdaptor. Both descriptor versions use the same StateTtlConfig class, and the V1 TTL wrappers implement every method the V2 state adaptors call.
> StateDescriptorUtils stays unchanged. Its other production caller, DefaultOperatorStateBackend, creates operator state, which does not support TTL.
> h2. Reproduction
> # Create an AsyncKeyedStateBackendAdaptor over a HeapKeyedStateBackend that uses a controllable TtlTimeProvider.
> # Create value, list, map, reducing and aggregating state from V2 descriptors with a 10 ms TTL.
> # Write one entry to each state at time 0, then advance the TTL time to 10 ms.
> # All five states still return their values.
> h2. Planned work
> I have prepared the one-line fix with a deterministic regression test covering all five state types. Two existing tests asserted the no-expiry behavior for async state on the HashMap backend: RowTimeDeduplicateFunctionTest and GroupAggregateHarnessTest#testAggregateWithRetraction. Their async expectations now match the sync ones. StreamingJoinOperatorTest skipped its four state retention tests for async state because of this issue; they now run and pass. No upstream PR has been opened yet.
> h2. Compatibility
> Existing async state jobs on HashMap or RocksDB with a configured TTL start expiring state after the upgrade.
> A checkpoint or savepoint written before the fix stores this state without TTL. Restoring it with TTL now enabled uses the existing TTL state migration of the HashMap and RocksDB backends, covered by StateBackendMigrationTestBase#testStateMigrationAfterChangingTTLFromDisablingToEnabling. Migrated entries are timestamped at restore time, so pre-existing state expires one TTL after the restore. Reducing and aggregating state use the same migration code as value state.
> Unlike the native ForSt path, the adaptor path applies the V1 TTL rules, including the V1 cleanup strategies and update types.
> Note on documentation: the State V2 page (docs/content/docs/dev/datastream/fault-tolerance/state_v2.md) says that restoring state with TTL enabled when it was written without TTL, or the reverse, fails with StateMigrationException. That describes native State V2 backends. On the adaptor path (HashMap and RocksDB) the V1 TTL migration applies instead. The proposed fix does not change the documentation; a clarification could follow separately.
> h2. Proposed release note
> Async state (the State V2 API, including SQL operators with table.exec.async-state.enabled) on the HashMap and RocksDB state backends now honors the configured state TTL. Previously such state never expired. After restoring from an earlier checkpoint or savepoint, existing entries are treated as written at restore time and expire one TTL later.



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)