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 <HASH>..<HASH> 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;
[FLINK-<I>] protect user accumulators against concurrent updates