[FLINK-4829] protect user accumulators against concurrent updates

shuttie · Oct 18, 2016 · 783dca5 · 783dca5
1 parent f46ca39
commit 783dca5
Showing 1 changed file with 3 additions and 1 deletion.
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
@@ -27,6 +27,7 @@
 
 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;