[FLINK-37463][state/forst] make serialzer in ForStMapState Threadlocal (#26286)

Author: mayuehappy

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java

@@ -44,7 +44,7 @@
 public class ForStDBBunchPutRequest<K, N, UK, UV> extends ForStDBPutRequest<K, N, Map<UK, UV>> {
 
     /** Serializer for the user values. */
-    final TypeSerializer<UV> userValueSerializer;
+    final ThreadLocal<TypeSerializer<UV>> userValueSerializer;
 
     /** The data outputStream used for value serializer, which should be thread-safe. */
     final ThreadLocal<DataOutputSerializer> valueSerializerView;
@@ -100,7 +100,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);
+        userValueSerializer.get().serialize(singleValue, outputView);
         return outputView.getCopyOfBuffer();
     }
 

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java

@@ -73,10 +73,10 @@ public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
     final ThreadLocal<DataInputDeserializer> valueDeserializerView;
 
     /** Serializer for the user keys. */
-    final TypeSerializer<UK> userKeySerializer;
+    final ThreadLocal<TypeSerializer<UK>> userKeySerializer;
 
     /** Serializer for the user values. */
-    final TypeSerializer<UV> userValueSerializer;
+    final ThreadLocal<TypeSerializer<UV>> userValueSerializer;
 
     /** Number of bytes required to prefix the key groups. */
     private final int keyGroupPrefixBytes;
@@ -100,8 +100,8 @@ public ForStMapState(
         this.valueSerializerView = ThreadLocal.withInitial(valueSerializerViewInitializer);
         this.keyDeserializerView = ThreadLocal.withInitial(keyDeserializerViewInitializer);
         this.valueDeserializerView = ThreadLocal.withInitial(valueDeserializerViewInitializer);
-        this.userKeySerializer = stateDescriptor.getUserKeySerializer();
-        this.userValueSerializer = stateDescriptor.getSerializer();
+        this.userKeySerializer = ThreadLocal.withInitial(stateDescriptor::getUserKeySerializer);
+        this.userValueSerializer = ThreadLocal.withInitial(stateDescriptor::getSerializer);
         this.keyGroupPrefixBytes = keyGroupPrefixBytes;
     }
 
@@ -124,15 +124,15 @@ public byte[] serializeKey(ContextKey<K, N> contextKey) throws IOException {
             return builder.build();
         }
         UK userKey = (UK) contextKey.getUserKey(); // map get
-        return builder.buildCompositeKeyUserKey(userKey, userKeySerializer);
+        return builder.buildCompositeKeyUserKey(userKey, userKeySerializer.get());
     }
 
     @Override
     public byte[] serializeValue(UV value) throws IOException {
         DataOutputSerializer outputView = valueSerializerView.get();
         outputView.clear();
         outputView.writeBoolean(false);
-        userValueSerializer.serialize(value, outputView);
+        userValueSerializer.get().serialize(value, outputView);
         return outputView.getCopyOfBuffer();
     }
 
@@ -141,13 +141,13 @@ public UV deserializeValue(byte[] valueBytes) throws IOException {
         DataInputDeserializer inputView = valueDeserializerView.get();
         inputView.setBuffer(valueBytes);
         boolean isNull = inputView.readBoolean();
-        return isNull ? null : userValueSerializer.deserialize(inputView);
+        return isNull ? null : userValueSerializer.get().deserialize(inputView);
     }
 
     public UK deserializeUserKey(byte[] userKeyBytes, int userKeyOffset) throws IOException {
         DataInputDeserializer inputView = keyDeserializerView.get();
         inputView.setBuffer(userKeyBytes, userKeyOffset, userKeyBytes.length - userKeyOffset);
-        return userKeySerializer.deserialize(inputView);
+        return userKeySerializer.get().deserialize(inputView);
     }
 
     @Override