flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From m..@apache.org
Subject [2/2] flink git commit: [FLINK-4829] protect user accumulators against concurrent updates
Date Tue, 18 Oct 2016 13:57:42 GMT
[FLINK-4829] protect user accumulators against concurrent updates


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

Branch: refs/heads/master
Commit: 783dca56eedc95f0a8974a9b50f2b532ca8cf849
Parents: f46ca39
Author: Maximilian Michels <mxm@apache.org>
Authored: Fri Oct 14 15:15:50 2016 +0200
Committer: Maximilian Michels <mxm@apache.org>
Committed: Tue Oct 18 15:55:33 2016 +0200

----------------------------------------------------------------------
 .../apache/flink/runtime/accumulators/AccumulatorRegistry.java   | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/783dca56/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
index 41af2a9..44714e7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 
 /**
@@ -44,7 +45,8 @@ public class AccumulatorRegistry {
 			new HashMap<Metric, Accumulator<?, ?>>();
 
 	/* User-defined Accumulator values stored for the executing task. */
-	private final Map<String, Accumulator<?, ?>> userAccumulators = new HashMap<>();
+	private final Map<String, Accumulator<?, ?>> userAccumulators =
+			new ConcurrentHashMap<>(4);
 
 	/* The reporter reference that is handed to the reporting tasks. */
 	private final ReadWriteReporter reporter;


Mime
View raw message