From 1523105915f48255aa22ac37fa72077b5f1fdbba Mon Sep 17 00:00:00 2001
From: Zakelly <zakelly.lan@gmail.com>
Date: Fri, 17 Jul 2026 00:20:49 +0800
Subject: [PATCH] [FLINK-40157][state/ForSt] Fix MapState putAll value
 serialization

---
 .../state/forst/ForStDBBunchPutRequest.java   | 21 +++----------------
 .../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 4a89fde78d30b..21713ca6186d3 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 @@
  */
 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 ForStDBBunchPutRequest(
             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 byte[] buildSerializedKey(UK userKey) throws IOException {
     }
 
     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 41abf364c383e..a340ae7db873f 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 @@ void testMapStateWriteBatch() throws Exception {
                     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();