fix: address CodeQL concurrency warnings

extensions-core/lookups-cached-global/src/main/java/org/apache/druid/server/lookup/namespace/cache/OffHeapNamespaceExtractionCacheManager.java

@@ -115,8 +115,10 @@ void disposeManually()
 
     private void doDispose()
     {
-      if (!mmapDB.isClosed()) {
-        mmapDB.delete(mapDbKey);
+      synchronized (mmapDB) {
+        if (!mmapDB.isClosed()) {
+          mmapDB.delete(mapDbKey);
+        }
       }
       cacheCount.decrementAndGet();
     }
@@ -150,10 +152,14 @@ protected ConcurrentMap<String, String> delegate()
     }
   }
 
+  /**
+   * MapDB synchronizes its state-changing methods on the {@link DB} instance. Check-and-use sequences must hold the
+   * same monitor to prevent the database from closing between the two calls.
+   */
   private final DB mmapDB;
   private final File tmpFile;
-  private AtomicLong mapDbKeyCounter = new AtomicLong(0);
-  private AtomicInteger cacheCount = new AtomicInteger(0);
+  private final AtomicLong mapDbKeyCounter = new AtomicLong(0);
+  private final AtomicInteger cacheCount = new AtomicInteger(0);
 
   @Inject
   public OffHeapNamespaceExtractionCacheManager(
@@ -192,14 +198,18 @@ public void start()
             }
 
             @Override
-            public synchronized void stop()
+            public void stop()
             {
-              if (!mmapDB.isClosed()) {
-                mmapDB.close();
-                if (!tmpFile.delete()) {
-                  log.warn("Unable to delete file at [%s]", tmpFile.getAbsolutePath());
+              final boolean shouldDelete;
+              synchronized (mmapDB) {
+                shouldDelete = !mmapDB.isClosed();
+                if (shouldDelete) {
+                  mmapDB.close();
                 }
               }
+              if (shouldDelete && !tmpFile.delete()) {
+                log.warn("Unable to delete file at [%s]", tmpFile.getAbsolutePath());
+              }
             }
           }
       );

processing/src/main/java/org/apache/druid/frame/channel/ReadableInputStreamFrameChannel.java

@@ -217,7 +217,7 @@ private void startReading()
                       () -> {
                         synchronized (readMonitor) {
                           keepReading = true;
-                          readMonitor.notify();
+                          readMonitor.notifyAll();
                         }
                       },
                       Execs.directExecutor()

processing/src/main/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStream.java

@@ -59,7 +59,7 @@ public void done()
   {
     synchronized (singleByteReaderDoer) {
       done = true;
-      singleByteReaderDoer.notify();
+      singleByteReaderDoer.notifyAll();
     }
   }
 
@@ -68,7 +68,7 @@ public void exceptionCaught(Throwable t)
     synchronized (singleByteReaderDoer) {
       done = true;
       throwable = t;
-      singleByteReaderDoer.notify();
+      singleByteReaderDoer.notifyAll();
     }
   }
 

processing/src/test/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStreamTest.java

@@ -257,4 +257,89 @@ public byte[] call()
     }
 
   }
+
+  @Test
+  public void testDoneUnblocksAllReaders() throws Exception
+  {
+    final AppendableByteArrayInputStream in = new AppendableByteArrayInputStream();
+    final AtomicReference<Integer> firstResult = new AtomicReference<>();
+    final AtomicReference<Integer> secondResult = new AtomicReference<>();
+    final AtomicReference<Throwable> firstError = new AtomicReference<>();
+    final AtomicReference<Throwable> secondError = new AtomicReference<>();
+    final Thread firstReader = readerThread(in, firstResult, firstError);
+    final Thread secondReader = readerThread(in, secondResult, secondError);
+
+    firstReader.start();
+    secondReader.start();
+    waitUntilWaiting(firstReader, secondReader);
+
+    in.done();
+
+    firstReader.join(1_000);
+    secondReader.join(1_000);
+    Assertions.assertFalse(firstReader.isAlive());
+    Assertions.assertFalse(secondReader.isAlive());
+    Assertions.assertNull(firstError.get());
+    Assertions.assertNull(secondError.get());
+    Assertions.assertEquals(-1, firstResult.get());
+    Assertions.assertEquals(-1, secondResult.get());
+  }
+
+  @Test
+  public void testExceptionUnblocksAllReaders() throws Exception
+  {
+    final AppendableByteArrayInputStream in = new AppendableByteArrayInputStream();
+    final AtomicReference<Integer> firstResult = new AtomicReference<>();
+    final AtomicReference<Integer> secondResult = new AtomicReference<>();
+    final AtomicReference<Throwable> firstError = new AtomicReference<>();
+    final AtomicReference<Throwable> secondError = new AtomicReference<>();
+    final Thread firstReader = readerThread(in, firstResult, firstError);
+    final Thread secondReader = readerThread(in, secondResult, secondError);
+
+    firstReader.start();
+    secondReader.start();
+    waitUntilWaiting(firstReader, secondReader);
+
+    final Exception expected = new Exception();
+    in.exceptionCaught(expected);
+
+    firstReader.join(1_000);
+    secondReader.join(1_000);
+    Assertions.assertFalse(firstReader.isAlive());
+    Assertions.assertFalse(secondReader.isAlive());
+    Assertions.assertNull(firstResult.get());
+    Assertions.assertNull(secondResult.get());
+    Assertions.assertSame(expected, firstError.get().getCause());
+    Assertions.assertSame(expected, secondError.get().getCause());
+  }
+
+  private static Thread readerThread(
+      final AppendableByteArrayInputStream in,
+      final AtomicReference<Integer> result,
+      final AtomicReference<Throwable> error
+  )
+  {
+    final Thread reader = new Thread(() -> {
+      try {
+        result.set(in.read());
+      }
+      catch (Throwable t) {
+        error.set(t);
+      }
+    });
+    reader.setDaemon(true);
+    return reader;
+  }
+
+  private static void waitUntilWaiting(final Thread... readers) throws InterruptedException
+  {
+    for (int i = 0; i < 100; i++) {
+      if (Arrays.stream(readers).allMatch(reader -> reader.getState() == Thread.State.WAITING)) {
+        return;
+      }
+      Thread.sleep(10);
+    }
+
+    Assertions.fail("Readers did not all block on the input stream");
+  }
 }