clean up some thread pools in tests (#17421)

processing/src/test/java/org/apache/druid/common/guava/GuavaUtilsTest.java

@@ -75,46 +75,53 @@ public void testCancelAll()
     int tasks = 3;
     ExecutorService service = Execs.multiThreaded(tasks, "GuavaUtilsTest-%d");
     ListeningExecutorService exc = MoreExecutors.listeningDecorator(service);
-    //a flag what time to throw exception.
-    AtomicBoolean someoneFailed = new AtomicBoolean(false);
-    List<CountDownLatch> latches = new ArrayList<>(tasks);
-    Function<Integer, List<ListenableFuture<Object>>> function = (taskCount) -> {
-      List<ListenableFuture<Object>> futures = new ArrayList<>();
-      for (int i = 0; i < taskCount; i++) {
-        final CountDownLatch latch = new CountDownLatch(1);
-        latches.add(latch);
-        ListenableFuture<Object> future = exc.submit(new Callable<Object>() {
-          @Override
-          public Object call() throws RuntimeException, InterruptedException
+    try {
+      //a flag what time to throw exception.
+      AtomicBoolean someoneFailed = new AtomicBoolean(false);
+      List<CountDownLatch> latches = new ArrayList<>(tasks);
+      Function<Integer, List<ListenableFuture<Object>>> function = (taskCount) -> {
+        List<ListenableFuture<Object>> futures = new ArrayList<>();
+        for (int i = 0; i < taskCount; i++) {
+          final CountDownLatch latch = new CountDownLatch(1);
+          latches.add(latch);
+          ListenableFuture<Object> future = exc.submit(new Callable<Object>()
           {
-            latch.await(60, TimeUnit.SECONDS);
-            if (someoneFailed.compareAndSet(false, true)) {
-              throw new RuntimeException("This exception simulates an error");
+            @Override
+            public Object call() throws RuntimeException, InterruptedException
+            {
+              latch.await(60, TimeUnit.SECONDS);
+              if (someoneFailed.compareAndSet(false, true)) {
+                throw new RuntimeException("This exception simulates an error");
+              }
+              return null;
             }
-            return null;
-          }
-        });
-        futures.add(future);
-      }
-      return futures;
-    };
+          });
+          futures.add(future);
+        }
+        return futures;
+      };
 
-    List<ListenableFuture<Object>> futures = function.apply(tasks);
-    Assert.assertEquals(tasks, futures.stream().filter(f -> !f.isDone()).count());
-    // "release" the last tasks, which will cause it to fail as someoneFailed will still be false
-    latches.get(tasks - 1).countDown();
+      List<ListenableFuture<Object>> futures = function.apply(tasks);
+      Assert.assertEquals(tasks, futures.stream().filter(f -> !f.isDone()).count());
+      // "release" the last tasks, which will cause it to fail as someoneFailed will still be false
+      latches.get(tasks - 1).countDown();
 
-    ListenableFuture<List<Object>> future = Futures.allAsList(futures);
+      ListenableFuture<List<Object>> future = Futures.allAsList(futures);
 
-    ExecutionException thrown = Assert.assertThrows(
-        ExecutionException.class,
-        future::get
-    );
-    Assert.assertEquals("This exception simulates an error", thrown.getCause().getMessage());
-    GuavaUtils.cancelAll(true, future, futures);
-    Assert.assertEquals(0, futures.stream().filter(f -> !f.isDone()).count());
-    for (CountDownLatch latch : latches) {
-      latch.countDown();
+      ExecutionException thrown = Assert.assertThrows(
+          ExecutionException.class,
+          future::get
+      );
+      Assert.assertEquals("This exception simulates an error", thrown.getCause().getMessage());
+      GuavaUtils.cancelAll(true, future, futures);
+      Assert.assertEquals(0, futures.stream().filter(f -> !f.isDone()).count());
+      for (CountDownLatch latch : latches) {
+        latch.countDown();
+      }
+    }
+    finally {
+      exc.shutdownNow();
+      service.shutdownNow();
     }
   }
 }

processing/src/test/java/org/apache/druid/java/util/emitter/core/HttpEmitterTest.java

@@ -72,21 +72,27 @@ public void timeoutEmptyQueue() throws IOException, InterruptedException
         .setHttpTimeoutAllowanceFactor(timeoutAllowanceFactor)
         .setFlushTimeout(BaseHttpEmittingConfig.TEST_FLUSH_TIMEOUT_MILLIS)
         .build();
-    final HttpPostEmitter emitter = new HttpPostEmitter(config, httpClient, OBJECT_MAPPER);
+    try (final HttpPostEmitter emitter = new HttpPostEmitter(config, httpClient, OBJECT_MAPPER)) {
+      long startMs = System.currentTimeMillis();
+      emitter.start();
+      emitter.emitAndReturnBatch(new IntEvent());
+      emitter.flush();
+      long fillTimeMs = System.currentTimeMillis() - startMs;
+      MatcherAssert.assertThat(
+          (double) timeoutUsed.get(),
+          Matchers.lessThan(fillTimeMs * (timeoutAllowanceFactor + 0.5))
+      );
 
-    long startMs = System.currentTimeMillis();
-    emitter.start();
-    emitter.emitAndReturnBatch(new IntEvent());
-    emitter.flush();
-    long fillTimeMs = System.currentTimeMillis() - startMs;
-    MatcherAssert.assertThat((double) timeoutUsed.get(), Matchers.lessThan(fillTimeMs * (timeoutAllowanceFactor + 0.5)));
-
-    startMs = System.currentTimeMillis();
-    final Batch batch = emitter.emitAndReturnBatch(new IntEvent());
-    Thread.sleep(1000);
-    batch.seal();
-    emitter.flush();
-    fillTimeMs = System.currentTimeMillis() - startMs;
-    MatcherAssert.assertThat((double) timeoutUsed.get(), Matchers.lessThan(fillTimeMs * (timeoutAllowanceFactor + 0.5)));
+      startMs = System.currentTimeMillis();
+      final Batch batch = emitter.emitAndReturnBatch(new IntEvent());
+      Thread.sleep(1000);
+      batch.seal();
+      emitter.flush();
+      fillTimeMs = System.currentTimeMillis() - startMs;
+      MatcherAssert.assertThat(
+          (double) timeoutUsed.get(),
+          Matchers.lessThan(fillTimeMs * (timeoutAllowanceFactor + 0.5))
+      );
+    }
   }
 }

processing/src/test/java/org/apache/druid/java/util/emitter/core/HttpPostEmitterLoggerStressTest.java

@@ -52,61 +52,63 @@ public void testBurstFollowedByQuietPeriod() throws InterruptedException, IOExce
         .setBatchQueueSizeLimit(10)
         .setMinHttpTimeoutMillis(100)
         .build();
-    final HttpPostEmitter emitter = new HttpPostEmitter(config, httpClient, new ObjectMapper());
+    try (HttpPostEmitter emitter = new HttpPostEmitter(config, httpClient, new ObjectMapper())) {
 
-    emitter.start();
+      emitter.start();
 
-    httpClient.setGoHandler(new GoHandler() {
-      @Override
-      protected ListenableFuture<Response> go(Request request)
+      httpClient.setGoHandler(new GoHandler()
       {
-        return GoHandlers.immediateFuture(EmitterTest.okResponse());
+        @Override
+        protected ListenableFuture<Response> go(Request request)
+        {
+          return GoHandlers.immediateFuture(EmitterTest.okResponse());
+        }
+      });
+
+      Event smallEvent = ServiceMetricEvent.builder()
+                                           .setFeed("smallEvents")
+                                           .setDimension("test", "hi")
+                                           .setMetric("metric", 10)
+                                           .build("qwerty", "asdfgh");
+
+      for (int i = 0; i < 1000; i++) {
+        emitter.emit(smallEvent);
+
+        Assert.assertTrue(emitter.getTotalFailedBuffers() <= 10);
+        Assert.assertTrue(emitter.getBuffersToEmit() <= 12);
       }
-    });
 
-    Event smallEvent = ServiceMetricEvent.builder()
-                                       .setFeed("smallEvents")
-                                       .setDimension("test", "hi")
-                                       .setMetric("metric", 10)
-                                       .build("qwerty", "asdfgh");
+      // by the end of this test, there should be no outstanding failed buffers
 
-    for (int i = 0; i < 1000; i++) {
-      emitter.emit(smallEvent);
+      // with a flush time of 5s, min timeout of 100ms, 20s should be
+      // easily enough to get through all of the events
 
-      Assert.assertTrue(emitter.getTotalFailedBuffers() <= 10);
-      Assert.assertTrue(emitter.getBuffersToEmit() <= 12);
-    }
-
-    // by the end of this test, there should be no outstanding failed buffers
-
-    // with a flush time of 5s, min timeout of 100ms, 20s should be
-    // easily enough to get through all of the events
+      while (emitter.getTotalFailedBuffers() > 0) {
+        Thread.sleep(500);
+      }
 
-    while (emitter.getTotalFailedBuffers() > 0) {
-      Thread.sleep(500);
+      // there is also no reason to have too many log events
+      // refer to: https://github.com/apache/druid/issues/11279;
+
+      long countOfTimeouts = logCapture.getLogEvents().stream()
+                                       .filter(ev -> ev.getLevel() == Level.DEBUG)
+                                       .filter(ev -> ev.getThrown() instanceof TimeoutException)
+                                       .count();
+
+      // 1000 events limit, implies we should have no more than
+      // 1000 rejected send events within the expected 20sec
+      // duration of the test
+      long limitTimeoutEvents = 1000;
+
+      Assert.assertTrue(
+          String.format(
+              Locale.getDefault(),
+              "too many timeouts (%d), expect less than (%d)",
+              countOfTimeouts,
+              limitTimeoutEvents
+          ),
+          countOfTimeouts < limitTimeoutEvents
+      );
     }
-
-    // there is also no reason to have too many log events
-    // refer to: https://github.com/apache/druid/issues/11279;
-
-    long countOfTimeouts = logCapture.getLogEvents().stream()
-        .filter(ev -> ev.getLevel() == Level.DEBUG)
-        .filter(ev -> ev.getThrown() instanceof TimeoutException)
-        .count();
-
-    // 1000 events limit, implies we should have no more than
-    // 1000 rejected send events within the expected 20sec
-    // duration of the test
-    long limitTimeoutEvents = 1000;
-
-    Assert.assertTrue(
-        String.format(
-          Locale.getDefault(),
-          "too many timeouts (%d), expect less than (%d)",
-          countOfTimeouts,
-          limitTimeoutEvents),
-        countOfTimeouts < limitTimeoutEvents);
-
-    emitter.close();
   }
 }