[GitHub] [flink] masteryhx commented on a diff in pull request #19679: [FLINK-23143][state/changelog] Support state migration for ChangelogS…

[GitBox <gi...@apache.org>]

masteryhx commented on code in PR #19679:
URL: https://github.com/apache/flink/pull/19679#discussion_r918587896


##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java:
##########
@@ -163,6 +170,32 @@ KeyGroupedInternalPriorityQueue<T> create(
         return priorityQueuesManager.createOrUpdate(stateName, byteOrderedElementSerializer);
     }
 
+    @Override
+    public <N, SV, SEV, S extends State, IS extends S> IS upgradeKeyedState(
+            TypeSerializer<N> namespaceSerializer,
+            StateDescriptor<S, SV> stateDescriptor,
+            @Nonnull StateSnapshotTransformFactory<SEV> snapshotTransformFactory)
+            throws Exception {
+        Preconditions.checkState(createdKVStates.containsKey(stateDescriptor.getName()));
+        registeredKVStates.computeIfPresent(
+                stateDescriptor.getName(),
+                (stateName, stateTable) -> {
+                    stateTable.setMetaInfo(
+                            new RegisteredKeyValueStateBackendMetaInfo<>(
+                                    stateTable.getMetaInfo().snapshot()));
+                    return stateTable;

Review Comment:
   1. Good point. I agree. BTW, It's a problem also exists In the original implement as you could see `stateTable.setMetaInfo ` in `tryRegisterStateTable`, right ? I'd like to split a seprate commit to solve it, WDYT ?
   2. Sure, I will add a comment.



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org