[FLINK-37294][state] Support state migration between disabling and enabling ttl in HeapKeyedStateBackend #26651

hejufang · 2025-06-08T09:17:24Z

[FLINK-37294][state] Support state migration between disabling and enabling ttl in HeapKeyedStateBackend

What is the purpose of the change

Support state migration between disabling and enabling ttl in HeapKeyedStateBackend

Brief change log

Add migrateTtlValue in AbstractHeapState. When the state TTL switch changes, trigger the migration of state data.

Verifying this change

This change is already covered by existing tests, such as StateBackendMigrationTestBase#testStateMigrationAfterChangingTTLFromEnablingToDisabling and StateBackendMigrationTestBase#testStateMigrationAfterChangingTTLFromDisablingToEnabling.

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: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2025-06-08T09:24:14Z

CI report:

40ae1e8 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2025-06-10T11:10:11Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/AbstractHeapState.java

+            TypeSerializer<SV> newSerializer,
+            TtlTimeProvider ttlTimeProvider) {
+
+        Preconditions.checkArgument(priorSerializer instanceof TtlAwareSerializer);


nit: in the variable names we use prior and new and previous and current. I suggest being consistent;
previousSerializer previousAwareSerializer
currentSerializer currentTtlAwareSerializer

davidradl · 2025-06-10T11:11:29Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

@@ -212,7 +214,7 @@ private <N, V> StateTable<K, N, V> tryRegisterStateTable(
            StateDescriptor<?, V> stateDesc,
            @Nonnull StateSnapshotTransformFactory<V> snapshotTransformFactory,
            boolean allowFutureMetadataUpdates)
-            throws StateMigrationException {
+            throws Exception {


why are we changing this away from putting out the specific exception - which seems best practise to me.

@davidradl migrateTtlAwareStateValues will also throw FlinkRuntimeException

I'd agree keep the original exception, as the FlinkRuntimeException is uncheck one and there is no need to claim throwing it.

davidradl · 2025-06-10T11:13:13Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

-                                + previousStateSerializer
-                                + ").");
+            } else if (stateCompatibility.isCompatibleAfterMigration()) {
+                migrateStateValues(stateDesc, previousStateSerializer, newStateSerializer);


what happens if the state is not compatible, I suggest we should at least log for that case - or should there be an error in that case?

davidradl · 2025-06-10T11:14:08Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

@@ -299,6 +292,56 @@ private <N, V> StateTable<K, N, V> tryRegisterStateTable(
        return stateTable;
    }

+    /** Only triggering state migration when the state TTL is turned on or off is supported. */


do we need the text is supported - I am not sure what this means.

davidradl · 2025-06-10T11:16:40Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java

+            TypeSerializer<List<V>> newSerializer,
+            TtlTimeProvider ttlTimeProvider) {
+
+        Preconditions.checkArgument(


it looks like there is suplication of code, can we re arrange the code so the following is not repeated for the different cases

priorSerializer instanceof TtlAwareSerializer.TtlAwareListSerializer); Preconditions.checkArgument( newSerializer instanceof TtlAwareSerializer.TtlAwareListSerializer); TtlAwareSerializer<V, ?> priorTtlAwareElementSerializer = ((TtlAwareSerializer.TtlAwareListSerializer<V>) priorSerializer) .getElementSerializer(); TtlAwareSerializer<V, ?> newTtlAwareElementSerializer = ((TtlAwareSerializer.TtlAwareListSerializer<V>) newSerializer) .getElementSerializer();

xiangyuf · 2025-06-12T16:24:58Z

@hejufang Hi, I did some improvement for your implementation in this PR(#26674). I've also changed this commit as a co-authored commit. PTAL.

…abling ttl in HeapKeyedStateBackend Co-authored-by: hejufang <hejufang@bytedance.com> Co-authored-by: Xiangyu Feng <xiangyu0xf@gmail.com>

hejufang · 2025-06-13T03:36:37Z

@hejufang Hi, I did some improvement for your implementation in this PR(#26674). I've also changed this commit as a co-authored commit. PTAL.

@xiangyuf Thank you for the improvements. I have push the new commit to the current branch, and we can continue to track this PR.

xiangyuf · 2025-06-13T03:42:58Z

Cool, I've closed another PR.

xiangyuf · 2025-06-13T06:28:38Z

@Zakelly Kindly remind for review.

xiangyuf · 2025-06-18T06:54:42Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

+                    "State should be an AbstractRocksDBState but is " + state);
+        }
+        AbstractHeapState<K, N, V> heapState = (AbstractHeapState<K, N, V>) state;
+        TtlAwareSerializer<V, ?> previousTtlAwareSerializer =


Variable previousTtlAwareSerializer is not used. You can safely delete this.

comment resolved

Zakelly

Thanks for the PR, overall looks good

Zakelly · 2025-06-18T15:56:26Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

@@ -212,7 +214,7 @@ private <N, V> StateTable<K, N, V> tryRegisterStateTable(
            StateDescriptor<?, V> stateDesc,
            @Nonnull StateSnapshotTransformFactory<V> snapshotTransformFactory,
            boolean allowFutureMetadataUpdates)
-            throws StateMigrationException {
+            throws Exception {


I'd agree keep the original exception, as the FlinkRuntimeException is uncheck one and there is no need to claim throwing it.

Zakelly · 2025-06-18T16:09:45Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

+
+        while (iterator.hasNext()) {
+            final StateEntry<K, N, V> entry = iterator.next();
+            stateTable.put(


It there any concurrency issue when performing stateTable.put() over the iteration?
If so, I'd suggest a new inner transform method to in-place update all the value of inner entries

hejufang force-pushed the FLINK-37294 branch from 7d80aba to d18e0cb Compare June 8, 2025 09:47

hejufang changed the title ~~[FLINK-37294][state] Support state migration between disabling and enabling ttl in RocksDBKeyedStateBackend~~ [FLINK-37294][state] Support state migration between disabling and enabling ttl in HeapKeyedStateBackend Jun 8, 2025

davidradl reviewed Jun 10, 2025

View reviewed changes

[FLINK-37294][state] Support state migration between disabling and en…

c6ccab7

…abling ttl in HeapKeyedStateBackend Co-authored-by: hejufang <hejufang@bytedance.com> Co-authored-by: Xiangyu Feng <xiangyu0xf@gmail.com>

hejufang force-pushed the FLINK-37294 branch from d18e0cb to c6ccab7 Compare June 13, 2025 03:25

xiangyuf reviewed Jun 18, 2025

View reviewed changes

fix comment

40ae1e8

Zakelly reviewed Jun 18, 2025

View reviewed changes

[FLINK-37294][state] Support state migration between disabling and enabling ttl in HeapKeyedStateBackend #26651

Are you sure you want to change the base?

[FLINK-37294][state] Support state migration between disabling and enabling ttl in HeapKeyedStateBackend #26651

Conversation

hejufang commented Jun 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jun 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xiangyuf commented Jun 12, 2025

Uh oh!

hejufang commented Jun 13, 2025

Uh oh!

xiangyuf commented Jun 13, 2025

Uh oh!

xiangyuf commented Jun 13, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Zakelly left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hejufang commented Jun 8, 2025 •

edited

Loading

flinkbot commented Jun 8, 2025 •

edited

Loading