[hotfix][state] Create new copy for each SerializedCompositeKeyBuilde…

…r in ForStStateBackend (apache#24816)
lzcmian · May 21, 2024 · 5a2c1e0 · 5a2c1e0
1 parent 11bd030
commit 5a2c1e0
Showing 1 changed file with 4 additions and 5 deletions.
diff --git a/...ckend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java b/...ckend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
@@ -113,15 +113,14 @@ public ForStKeyedStateBackend<K> build() throws BackendBuildingException {
         int keyGroupPrefixBytes =
                 CompositeKeySerializationUtils.computeRequiredBytesInKeyGroupPrefix(
                         numberOfKeyGroups);
-        // it is important that we only create the key builder after the restore, and not
-        // before;
-        // restore operations may reconfigure the key serializer, so accessing the key
-        // serializer
+        // it is important that we only create the key builder after the restore, and not before;
+        // restore operations may reconfigure the key serializer, so accessing the key serializer
         // only now we can be certain that the key serializer used in the builder is final.
         Supplier<SerializedCompositeKeyBuilder<K>> serializedKeyBuilder =
                 () ->
                         new SerializedCompositeKeyBuilder<>(
-                                keySerializerProvider.currentSchemaSerializer(),
+                                // must create new copy for each SerializedCompositeKeyBuilder
+                                keySerializerProvider.currentSchemaSerializer().duplicate(),
                                 keyGroupPrefixBytes,
                                 KEY_SERIALIZER_BUFFER_START_SIZE);
         Supplier<DataOutputSerializer> valueSerializerView =