flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject flink git commit: [FLINK-3450] Duplicate TypeSerializer in StateDescriptor.writeObject
Date Mon, 22 Feb 2016 14:00:09 GMT
Repository: flink
Updated Branches:
  refs/heads/master 80c0c65b7 -> bfc14ebc1


[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".


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bfc14ebc
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bfc14ebc
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bfc14ebc

Branch: refs/heads/master
Commit: bfc14ebc1e7d961095896a0230512be59b13537b
Parents: 80c0c65
Author: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Authored: Fri Feb 19 15:38:57 2016 +0100
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Mon Feb 22 14:59:22 2016 +0100

----------------------------------------------------------------------
 .../java/org/apache/flink/api/common/state/StateDescriptor.java  | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/bfc14ebc/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java
----------------------------------------------------------------------
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 ab625cf..10ac5ba 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 @@ public abstract class StateDescriptor<S extends State, T> implements
Serializabl
 			try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); 
 					DataOutputViewStreamWrapper outView = new DataOutputViewStreamWrapper(baos))
 			{
-				serializer.serialize(defaultValue, outView);
+				TypeSerializer<T> duplicateSerializer = serializer.duplicate();
+				duplicateSerializer.serialize(defaultValue, outView);
+
 				outView.flush();
 				serializedDefaultValue = baos.toByteArray();
 			}


Mime
View raw message