[
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)
