Aaron He created FLINK-41093:
--------------------------------
Summary: 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
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)
