Prohibit assigning concurrent maps into Map-typed variables and fields and fix a race condition in CoordinatorRuleManager

benchmarks/src/main/java/org/apache/druid/benchmark/FilteredAggregatorBenchmark.java

@@ -201,7 +201,7 @@ public void setup() throws IOException
 
     factory = new TimeseriesQueryRunnerFactory(
         new TimeseriesQueryQueryToolChest(
-            QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
         ),
         new TimeseriesQueryEngine(),
         QueryBenchmarkUtil.NOOP_QUERYWATCHER

benchmarks/src/main/java/org/apache/druid/benchmark/GroupByTypeInterfaceBenchmark.java

@@ -407,7 +407,7 @@ public String getFormatString()
         strategySelector,
         new GroupByQueryQueryToolChest(
             strategySelector,
-            QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
         )
     );
   }

benchmarks/src/main/java/org/apache/druid/benchmark/TopNTypeInterfaceBenchmark.java

@@ -299,7 +299,10 @@ public void setup() throws IOException
             0,
             Integer.MAX_VALUE
         ),
-        new TopNQueryQueryToolChest(new TopNQueryConfig(), QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()),
+        new TopNQueryQueryToolChest(
+            new TopNQueryConfig(),
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
+        ),
         QueryBenchmarkUtil.NOOP_QUERYWATCHER
     );
   }

benchmarks/src/main/java/org/apache/druid/benchmark/query/GroupByBenchmark.java

@@ -495,7 +495,7 @@ public String getFormatString()
         strategySelector,
         new GroupByQueryQueryToolChest(
             strategySelector,
-            QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
         )
     );
   }

benchmarks/src/main/java/org/apache/druid/benchmark/query/QueryBenchmarkUtil.java

@@ -49,7 +49,7 @@ public static <T, QueryType extends Query<T>> QueryRunner<T> makeQueryRunner(
     );
   }
 
-  public static IntervalChunkingQueryRunnerDecorator NoopIntervalChunkingQueryRunnerDecorator()
+  public static IntervalChunkingQueryRunnerDecorator noopIntervalChunkingQueryRunnerDecorator()
   {
     return new IntervalChunkingQueryRunnerDecorator(null, null, null) {
       @Override

benchmarks/src/main/java/org/apache/druid/benchmark/query/SearchBenchmark.java

@@ -373,7 +373,7 @@ public void setup() throws IOException
         new SearchStrategySelector(Suppliers.ofInstance(config)),
         new SearchQueryQueryToolChest(
             config,
-            QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
         ),
         QueryBenchmarkUtil.NOOP_QUERYWATCHER
     );

benchmarks/src/main/java/org/apache/druid/benchmark/query/SelectBenchmark.java

@@ -233,7 +233,7 @@ public void setup() throws IOException
     factory = new SelectQueryRunnerFactory(
         new SelectQueryQueryToolChest(
             JSON_MAPPER,
-            QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator(),
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator(),
             selectConfigSupplier
         ),
         new SelectQueryEngine(),

benchmarks/src/main/java/org/apache/druid/benchmark/query/TimeseriesBenchmark.java

@@ -296,7 +296,7 @@ public void setup() throws IOException
 
     factory = new TimeseriesQueryRunnerFactory(
         new TimeseriesQueryQueryToolChest(
-            QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
         ),
         new TimeseriesQueryEngine(),
         QueryBenchmarkUtil.NOOP_QUERYWATCHER

benchmarks/src/main/java/org/apache/druid/benchmark/query/TopNBenchmark.java

@@ -278,7 +278,10 @@ public void setup() throws IOException
             0,
             Integer.MAX_VALUE
         ),
-        new TopNQueryQueryToolChest(new TopNQueryConfig(), QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()),
+        new TopNQueryQueryToolChest(
+            new TopNQueryConfig(),
+            QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
+        ),
         QueryBenchmarkUtil.NOOP_QUERYWATCHER
     );
   }

benchmarks/src/main/java/org/apache/druid/benchmark/query/timecompare/TimeCompareBenchmark.java

@@ -229,7 +229,10 @@ private void setupQueries()
               0,
               Integer.MAX_VALUE
           ),
-          new TopNQueryQueryToolChest(new TopNQueryConfig(), QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()),
+          new TopNQueryQueryToolChest(
+              new TopNQueryConfig(),
+              QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()
+          ),
           QueryBenchmarkUtil.NOOP_QUERYWATCHER
       );
     }
@@ -270,7 +273,7 @@ private void setupQueries()
 
       timeseriesQuery = timeseriesQueryBuilder.build();
       timeseriesFactory = new TimeseriesQueryRunnerFactory(
-          new TimeseriesQueryQueryToolChest(QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()),
+          new TimeseriesQueryQueryToolChest(QueryBenchmarkUtil.noopIntervalChunkingQueryRunnerDecorator()),
           new TimeseriesQueryEngine(),
           QueryBenchmarkUtil.NOOP_QUERYWATCHER
       );

core/src/main/java/org/apache/druid/java/util/emitter/core/ParametrizedUriEmitter.java

@@ -120,21 +120,19 @@ public void emit(Event event)
   {
     try {
       URI uri = uriExtractor.apply(event);
+      // get() before computeIfAbsent() is an optimization to avoid locking in computeIfAbsent() if not needed.
+      // See https://github.com/apache/incubator-druid/pull/6898#discussion_r251384586.
       HttpPostEmitter emitter = emitters.get(uri);
       if (emitter == null) {
         try {
           emitter = emitters.computeIfAbsent(uri, u -> {
             try {
               return innerLifecycle.addMaybeStartManagedInstance(
-                  new HttpPostEmitter(
-                      config.buildHttpEmitterConfig(u.toString()),
-                      client,
-                      jsonMapper
-                  )
+                  new HttpPostEmitter(config.buildHttpEmitterConfig(u.toString()), client, jsonMapper)
               );
             }
             catch (Exception e) {
-              throw Throwables.propagate(e);
+              throw new RuntimeException(e);
             }
           });
         }

core/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java

@@ -24,13 +24,16 @@
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
-import com.google.common.collect.ImmutableSet;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.io.Closer;
 import org.apache.druid.java.util.common.logger.Logger;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.util.ArrayDeque;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -123,13 +126,23 @@ protected void finalize() throws Throwable
   public void close()
   {
     closed.set(true);
-    final Map<K, ImmediateCreationResourceHolder<K, V>> mapView = pool.asMap();
-    for (K k : ImmutableSet.copyOf(mapView.keySet())) {
-      mapView.remove(k).close();
+    final ConcurrentMap<K, ImmediateCreationResourceHolder<K, V>> mapView = pool.asMap();
+    Closer closer = Closer.create();
+    for (Iterator<Map.Entry<K, ImmediateCreationResourceHolder<K, V>>> iterator =
+         mapView.entrySet().iterator(); iterator.hasNext(); ) {
+      Map.Entry<K, ImmediateCreationResourceHolder<K, V>> e = iterator.next();
+      iterator.remove();
+      closer.register(e.getValue());
+    }
+    try {
+      closer.close();
+    }
+    catch (IOException e) {
+      throw new RuntimeException(e);
     }
   }
 
-  private static class ImmediateCreationResourceHolder<K, V>
+  private static class ImmediateCreationResourceHolder<K, V> implements Closeable
   {
     private final int maxSize;
     private final K key;
@@ -265,7 +278,8 @@ private boolean holderListContains(V object)
       return resourceHolderList.stream().anyMatch(a -> a.getResource().equals(object));
     }
 
-    void close()
+    @Override
+    public void close()
     {
       synchronized (this) {
         closed = true;

extensions-contrib/materialized-view-selection/src/test/java/org/apache/druid/query/materializedview/MaterializedViewQueryQueryToolChestTest.java

@@ -58,7 +58,7 @@ public void testMakePostComputeManipulatorFn()
                 .put(
                     TimeseriesQuery.class,
                     new TimeseriesQueryQueryToolChest(
-                        QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+                        QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
                     )
                 )
                 .build()

extensions-contrib/virtual-columns/src/test/java/org/apache/druid/segment/MapVirtualColumnGroupByTest.java

@@ -109,7 +109,7 @@ public int getNumThreads()
         strategySelector,
         new GroupByQueryQueryToolChest(
             strategySelector,
-            QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
         )
     );
 

extensions-contrib/virtual-columns/src/test/java/org/apache/druid/segment/MapVirtualColumnSelectTest.java

@@ -70,7 +70,7 @@ public static Iterable<Object[]> constructorFeeder() throws IOException
     SelectQueryRunnerFactory factory = new SelectQueryRunnerFactory(
         new SelectQueryQueryToolChest(
             new DefaultObjectMapper(),
-            QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator(),
+            QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator(),
             selectConfigSupplier
         ),
         new SelectQueryEngine(),

extensions-contrib/virtual-columns/src/test/java/org/apache/druid/segment/MapVirtualColumnTopNTest.java

@@ -69,7 +69,7 @@ public void setup() throws IOException
         new StupidPool<>("map-virtual-column-test", () -> ByteBuffer.allocate(1024)),
         new TopNQueryQueryToolChest(
             new TopNQueryConfig(),
-            QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+            QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
         ),
         QueryRunnerTestHelper.NOOP_QUERYWATCHER
     );

extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/ApproximateHistogramTopNQueryTest.java

@@ -81,7 +81,7 @@ public static Iterable<Object[]> constructorFeeder()
                     defaultPool,
                     new TopNQueryQueryToolChest(
                         new TopNQueryConfig(),
-                        QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+                        QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
                     ),
                     QueryRunnerTestHelper.NOOP_QUERYWATCHER
                 )
@@ -91,7 +91,7 @@ public static Iterable<Object[]> constructorFeeder()
                     customPool,
                     new TopNQueryQueryToolChest(
                         new TopNQueryConfig(),
-                        QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+                        QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
                     ),
                     QueryRunnerTestHelper.NOOP_QUERYWATCHER
                 )

extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTopNQueryTest.java

@@ -81,7 +81,7 @@ public static Iterable<Object[]> constructorFeeder()
                     defaultPool,
                     new TopNQueryQueryToolChest(
                         new TopNQueryConfig(),
-                        QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+                        QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
                     ),
                     QueryRunnerTestHelper.NOOP_QUERYWATCHER
                 )
@@ -91,7 +91,7 @@ public static Iterable<Object[]> constructorFeeder()
                     customPool,
                     new TopNQueryQueryToolChest(
                         new TopNQueryConfig(),
-                        QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+                        QueryRunnerTestHelper.noopIntervalChunkingQueryRunnerDecorator()
                     ),
                     QueryRunnerTestHelper.NOOP_QUERYWATCHER
                 )

extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java

@@ -54,6 +54,7 @@
 import java.util.Properties;
 import java.util.UUID;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -184,7 +185,7 @@ public boolean start()
       final String topic = getKafkaTopic();
       LOG.debug("About to listen to topic [%s] with group.id [%s]", topic, factoryId);
       cacheHandler = cacheManager.createCache();
-      final Map<String, String> map = cacheHandler.getCache();
+      final ConcurrentMap<String, String> map = cacheHandler.getCache();
       mapRef.set(map);
       // Enable publish-subscribe
       kafkaProperties.setProperty("auto.offset.reset", "smallest");

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/LegacyKafkaIndexTaskRunner.java

@@ -104,6 +104,7 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -123,8 +124,8 @@ public class LegacyKafkaIndexTaskRunner extends SeekableStreamIndexTaskRunner<In
   private static final EmittingLogger log = new EmittingLogger(LegacyKafkaIndexTaskRunner.class);
   private static final String METADATA_NEXT_PARTITIONS = "nextPartitions";
 
-  private final Map<Integer, Long> endOffsets = new ConcurrentHashMap<>();
-  private final Map<Integer, Long> nextOffsets = new ConcurrentHashMap<>();
+  private final ConcurrentMap<Integer, Long> endOffsets = new ConcurrentHashMap<>();
+  private final ConcurrentMap<Integer, Long> nextOffsets = new ConcurrentHashMap<>();
 
   // The pause lock and associated conditions are to support coordination between the Jetty threads and the main
   // ingestion loop. The goal is to provide callers of the API a guarantee that if pause() returns successfully

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -50,7 +50,6 @@
 import org.apache.druid.discovery.LookupNodeService;
 import org.apache.druid.indexer.TaskState;
 import org.apache.druid.indexer.TaskStatus;
-import org.apache.druid.indexing.common.Counters;
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
 import org.apache.druid.indexing.common.SegmentLoaderFactory;
 import org.apache.druid.indexing.common.TaskLock;
@@ -2393,8 +2392,7 @@ public boolean checkPointDataSourceMetadata(
             );
             return true;
           }
-        },
-        new Counters()
+        }
     );
     final TaskActionClientFactory taskActionClientFactory = new LocalTaskActionClientFactory(
         taskStorage,

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java

@@ -68,6 +68,7 @@
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
@@ -341,7 +342,8 @@ private void rescheduleRunnable(long delayMillis)
 
   private ScheduledExecutorService scheduledExec;
 
-  private final Map<StreamPartition<String>, PartitionResource> partitionResources = new ConcurrentHashMap<>();
+  private final ConcurrentMap<StreamPartition<String>, PartitionResource> partitionResources =
+      new ConcurrentHashMap<>();
   private BlockingQueue<OrderedPartitionableRecord<String, String>> records;
 
   private volatile boolean checkPartitionsStarted = false;

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java

@@ -55,7 +55,6 @@
 import org.apache.druid.discovery.LookupNodeService;
 import org.apache.druid.indexer.TaskState;
 import org.apache.druid.indexer.TaskStatus;
-import org.apache.druid.indexing.common.Counters;
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
 import org.apache.druid.indexing.common.SegmentLoaderFactory;
 import org.apache.druid.indexing.common.TaskLock;
@@ -2705,8 +2704,7 @@ public boolean checkPointDataSourceMetadata(
             );
             return true;
           }
-        },
-        new Counters()
+        }
     );
     final TaskActionClientFactory taskActionClientFactory = new LocalTaskActionClientFactory(
         taskStorage,