From 51d8dc9fa5cdd84b6cc1c7711af96ddabf12aaf6 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Fri, 19 Feb 2016 15:38:57 +0100 Subject: [PATCH] [FLINK-3450] Duplicate TypeSerializer in StateDescriptor.writeObject The StateDescriptor can be serializer asynchronously in case of asynchronous checkpoints. In that case two threads would try to concurrently use the TypeSerializer: The normal state updating and the checkpoint serialization. If the TypeSerializer is a KryoSerializer this can lead to problems. Therefore the need to duplicate it before using in "writeObject". --- .../org/apache/flink/api/common/state/StateDescriptor.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java b/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java index ab625cfaf77d7a..10ac5bae0994df 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java @@ -260,7 +260,9 @@ private void writeObject(final ObjectOutputStream out) throws IOException { try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputViewStreamWrapper outView = new DataOutputViewStreamWrapper(baos)) { - serializer.serialize(defaultValue, outView); + TypeSerializer duplicateSerializer = serializer.duplicate(); + duplicateSerializer.serialize(defaultValue, outView); + outView.flush(); serializedDefaultValue = baos.toByteArray(); }