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

[zakelly]

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

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


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

commit d6ae4b9e5272b8b64a7bb6623719d69af665ceda
Author: Zakelly <zakelly....@gmail.com>
AuthorDate: Wed Jul 22 00:18:17 2026 +0800

    [FLINK-40157][state/ForSt] Fix MapState putAll value serialization (#28771)
---
 .../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 4a89fde78d3..21713ca6186 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,9 +19,6 @@
 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.memory.DataInputDeserializer;
-import org.apache.flink.core.memory.DataOutputSerializer;
 import org.apache.flink.core.state.InternalStateFuture;
 
 import org.forstdb.RocksDB;
@@ -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 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,
             InternalStateFuture<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.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 41abf364c38..a340ae7db87 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();