You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2026/07/21 16:18:21 UTC
(flink) branch release-2.1 updated: [FLINK-40157][state/ForSt] Fix MapState putAll value serialization (#28772)
This is an automated email from the ASF dual-hosted git repository.
Zakelly pushed a commit to branch release-2.1
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.1 by this push:
new bfca945efb4 [FLINK-40157][state/ForSt] Fix MapState putAll value serialization (#28772)
bfca945efb4 is described below
commit bfca945efb45b3386e328ac2a68f0d3ac046b8fb
Author: Zakelly <za...@gmail.com>
AuthorDate: Wed Jul 22 00:18:13 2026 +0800
[FLINK-40157][state/ForSt] Fix MapState putAll value serialization (#28772)
---
.../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();