(flink) branch master updated: [FLINK-40157][state/ForSt] Fix MapState putAll value serialization (#28762)

[zakelly]

This is an automated email from the ASF dual-hosted git repository.

Zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 642673e8d4e [FLINK-40157][state/ForSt] Fix MapState putAll value 
serialization (#28762)
642673e8d4e is described below

commit 642673e8d4e473a5a282018d08ba1416a850e1ec
Author: Zakelly <zakelly....@gmail.com>
AuthorDate: Fri Jul 17 20:07:11 2026 +0800

    [FLINK-40157][state/ForSt] Fix MapState putAll value serialization (#28762)
---
 .../flink/state/forst/ForStDBBunchPutRequest.java   | 21 +++------------------
 .../state/forst/ForStWriteBatchOperationTest.java   |  1 +
 2 files changed, 4 insertions(+), 18 deletions(-)

diff --git 
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
 
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
index 1fb3e5f7962..8cbbbfb05b3 100644
--- 
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
+++ 
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
@@ -19,10 +19,7 @@
 package org.apache.flink.state.forst;
 
 import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.core.asyncprocessing.InternalAsyncFuture;
-import org.apache.flink.core.memory.DataInputDeserializer;
-import org.apache.flink.core.memory.DataOutputSerializer;
 
 import org.forstdb.RocksDB;
 import org.forstdb.RocksDBException;
@@ -43,14 +40,7 @@ import static 
org.apache.flink.state.forst.ForStDBIterRequest.startWithKeyPrefix
  */
 public class ForStDBBunchPutRequest<K, N, UK, UV> extends ForStDBPutRequest<K, 
N, Map<UK, UV>> {
 
-    /** Serializer for the user values. */
-    final ThreadLocal<TypeSerializer<UV>> userValueSerializer;
-
-    /** The data outputStream used for value serializer, which should be 
thread-safe. */
-    final ThreadLocal<DataOutputSerializer> valueSerializerView;
-
-    /** The data inputStream used for value deserializer, which should be 
thread-safe. */
-    final ThreadLocal<DataInputDeserializer> valueDeserializerView;
+    private final ForStMapState<K, N, UK, UV> mapState;
 
     final int keyGroupPrefixBytes;
 
@@ -60,9 +50,7 @@ public class ForStDBBunchPutRequest<K, N, UK, UV> extends 
ForStDBPutRequest<K, N
             ForStMapState<K, N, UK, UV> table,
             InternalAsyncFuture<Void> future) {
         super(key, value, false, (ForStInnerTable<K, N, Map<UK, UV>>) table, 
future);
-        this.userValueSerializer = table.userValueSerializer;
-        this.valueSerializerView = table.valueSerializerView;
-        this.valueDeserializerView = table.valueDeserializerView;
+        this.mapState = table;
         this.keyGroupPrefixBytes = table.getKeyGroupPrefixBytes();
     }
 
@@ -98,10 +86,7 @@ public class ForStDBBunchPutRequest<K, N, UK, UV> extends 
ForStDBPutRequest<K, N
     }
 
     public byte[] buildSerializedValue(UV singleValue) throws IOException {
-        DataOutputSerializer outputView = valueSerializerView.get();
-        outputView.clear();
-        userValueSerializer.get().serialize(singleValue, outputView);
-        return outputView.getCopyOfBuffer();
+        return mapState.serializeValue(singleValue);
     }
 
     // --------------- For testing usage ---------------
diff --git 
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStWriteBatchOperationTest.java
 
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStWriteBatchOperationTest.java
index 6aa296e86da..0aeafff7efe 100644
--- 
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStWriteBatchOperationTest.java
+++ 
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStWriteBatchOperationTest.java
@@ -244,6 +244,7 @@ public class ForStWriteBatchOperationTest extends 
ForStDBOperationTestBase {
                     byte[] valueBytes = 
db.get(request.getColumnFamilyHandle(), keyBytes);
                     assertArrayEquals(
                             valueBytes, 
bunchPutRequest.buildSerializedValue(en.getValue()));
+                    
assertThat(mapState1.deserializeValue(valueBytes)).isEqualTo(en.getValue());
                 }
             } else {
                 byte[] keyBytes = request.buildSerializedKey();