Update HashMaps to ConcurrentHashMaps to resolve ConcurrentModificati… (

#496)
wavefrontHQ · Feb 13, 2020 · 1571c89 · 1571c89
1 parent d819821
commit 1571c89
Show file tree

Hide file tree

Showing 6 changed files with 23 additions and 19 deletions.
diff --git a/proxy/src/main/java/com/wavefront/agent/PushAgent.java b/proxy/src/main/java/com/wavefront/agent/PushAgent.java
@@ -114,6 +114,7 @@
 import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -678,7 +679,8 @@ protected void startDeltaCounterListener(String strPort,
 
     if (this.deltaCounterHandlerFactory == null) {
       this.deltaCounterHandlerFactory = new ReportableEntityHandlerFactory() {
-        private final Map<String, ReportableEntityHandler<?, ?>> handlers = new HashMap<>();
+        private final Map<String, ReportableEntityHandler<?, ?>> handlers =
+            new ConcurrentHashMap<>();
 
         @Override
         public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) {
@@ -935,7 +937,8 @@ protected void startHistogramListeners(List<String> ports,
     });
 
     ReportableEntityHandlerFactory histogramHandlerFactory = new ReportableEntityHandlerFactory() {
-      private final Map<HandlerKey, ReportableEntityHandler<?, ?>> handlers = new HashMap<>();
+      private final Map<HandlerKey, ReportableEntityHandler<?, ?>> handlers =
+          new ConcurrentHashMap<>();
       @SuppressWarnings("unchecked")
       @Override
       public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) {

diff --git a/proxy/src/main/java/com/wavefront/agent/handlers/ReportableEntityHandlerFactoryImpl.java b/proxy/src/main/java/com/wavefront/agent/handlers/ReportableEntityHandlerFactoryImpl.java
@@ -6,8 +6,8 @@
 import org.apache.commons.lang.math.NumberUtils;
 import wavefront.report.Histogram;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.logging.Logger;
 
@@ -43,7 +43,7 @@ public class ReportableEntityHandlerFactoryImpl implements ReportableEntityHandl
       getSystemPropertyAsDouble("wavefront.proxy.logevents.sample-rate"), false, logger::info);
 
   protected final Map<String, Map<ReportableEntityType, ReportableEntityHandler<?, ?>>> handlers =
-      new HashMap<>();
+      new ConcurrentHashMap<>();
 
   private final SenderTaskFactory senderTaskFactory;
   private final int blockedItemsPerBatch;
@@ -80,7 +80,7 @@ public ReportableEntityHandlerFactoryImpl(
   @Override
   public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) {
     return (ReportableEntityHandler<T, U>) handlers.computeIfAbsent(handlerKey.getHandle(),
-        h -> new HashMap<>()).computeIfAbsent(handlerKey.getEntityType(), k -> {
+        h -> new ConcurrentHashMap<>()).computeIfAbsent(handlerKey.getEntityType(), k -> {
       switch (handlerKey.getEntityType()) {
         case POINT:
           return new ReportPointHandlerImpl(handlerKey, blockedItemsPerBatch,

diff --git a/proxy/src/main/java/com/wavefront/agent/handlers/SenderTaskFactoryImpl.java b/proxy/src/main/java/com/wavefront/agent/handlers/SenderTaskFactoryImpl.java
@@ -17,11 +17,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -42,15 +42,15 @@
  */
 public class SenderTaskFactoryImpl implements SenderTaskFactory {
 
-  private final Map<String, List<ReportableEntityType>> entityTypes = new HashMap<>();
-  private final Map<HandlerKey, ScheduledExecutorService> executors = new HashMap<>();
-  private final Map<HandlerKey, List<SenderTask<?>>> managedTasks = new HashMap<>();
-  private final Map<HandlerKey, Managed> managedServices = new HashMap<>();
+  private final Map<String, List<ReportableEntityType>> entityTypes = new ConcurrentHashMap<>();
+  private final Map<HandlerKey, ScheduledExecutorService> executors = new ConcurrentHashMap<>();
+  private final Map<HandlerKey, List<SenderTask<?>>> managedTasks = new ConcurrentHashMap<>();
+  private final Map<HandlerKey, Managed> managedServices = new ConcurrentHashMap<>();
 
   /**
    * Keep track of all {@link TaskSizeEstimator} instances to calculate global buffer fill rate.
    */
-  private final Map<HandlerKey, TaskSizeEstimator> taskSizeEstimators = new HashMap<>();
+  private final Map<HandlerKey, TaskSizeEstimator> taskSizeEstimators = new ConcurrentHashMap<>();
 
   private final APIContainer apiContainer;
   private final UUID proxyId;

diff --git a/proxy/src/main/java/com/wavefront/agent/queueing/QueueingFactoryImpl.java b/proxy/src/main/java/com/wavefront/agent/queueing/QueueingFactoryImpl.java
@@ -13,11 +13,11 @@
 import com.wavefront.data.ReportableEntityType;
 
 import javax.annotation.Nonnull;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
@@ -30,9 +30,10 @@
  */
 public class QueueingFactoryImpl implements QueueingFactory {
 
-  private final Map<HandlerKey, ScheduledExecutorService> executors = new HashMap<>();
-  private final Map<HandlerKey, Map<Integer, QueueProcessor<?>>> queueProcessors = new HashMap<>();
-  private final Map<HandlerKey, QueueController<?>> queueControllers = new HashMap<>();
+  private final Map<HandlerKey, ScheduledExecutorService> executors = new ConcurrentHashMap<>();
+  private final Map<HandlerKey, Map<Integer, QueueProcessor<?>>> queueProcessors =
+      new ConcurrentHashMap<>();
+  private final Map<HandlerKey, QueueController<?>> queueControllers = new ConcurrentHashMap<>();
   private final TaskQueueFactory taskQueueFactory;
   private final APIContainer apiContainer;
   private final UUID proxyId;

diff --git a/proxy/src/main/java/com/wavefront/agent/queueing/TaskQueueFactoryImpl.java b/proxy/src/main/java/com/wavefront/agent/queueing/TaskQueueFactoryImpl.java
@@ -13,10 +13,10 @@
 import java.io.File;
 import java.io.RandomAccessFile;
 import java.nio.channels.FileChannel;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Logger;
 
 /**
@@ -27,7 +27,7 @@
 public class TaskQueueFactoryImpl implements TaskQueueFactory {
   private static final Logger logger =
       Logger.getLogger(TaskQueueFactoryImpl.class.getCanonicalName());
-  private final Map<HandlerKey, Map<Integer, TaskQueue<?>>> taskQueues = new HashMap<>();
+  private final Map<HandlerKey, Map<Integer, TaskQueue<?>>> taskQueues = new ConcurrentHashMap<>();
 
   private final String bufferFile;
   private final boolean purgeBuffer;

diff --git a/proxy/src/main/java/com/wavefront/common/SharedRateLimitingLogger.java b/proxy/src/main/java/com/wavefront/common/SharedRateLimitingLogger.java
@@ -2,8 +2,8 @@
 
 import com.google.common.util.concurrent.RateLimiter;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.LogRecord;
@@ -15,7 +15,7 @@
  */
 @SuppressWarnings("UnstableApiUsage")
 public class SharedRateLimitingLogger extends DelegatingLogger {
-  private static final Map<String, RateLimiter> SHARED_CACHE = new HashMap<>();
+  private static final Map<String, RateLimiter> SHARED_CACHE = new ConcurrentHashMap<>();
 
   private final RateLimiter rateLimiter;