Fix some ExecutorService leaks and use @cleanup("shutdownNow") for cleanup (#10198)

- use ExecutorService.shutdownNow() instead of ExecutorService.shutdown() in tests

Author: lhotari

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

@@ -60,6 +60,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import lombok.Cleanup;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.BookKeeper.DigestType;
@@ -708,6 +709,7 @@ void testConcurrentResetCursor() throws Exception {
         final int Consumers = 5;
 
         List<Future<AtomicBoolean>> futures = Lists.newArrayList();
+        @Cleanup("shutdownNow")
         ExecutorService executor = Executors.newCachedThreadPool();
         final CyclicBarrier barrier = new CyclicBarrier(Consumers + 1);
 
@@ -1727,6 +1729,7 @@ void testReadEntriesOrWaitBlocking() throws Exception {
         final int Consumers = 10;
 
         List<Future<Void>> futures = Lists.newArrayList();
+        @Cleanup("shutdownNow")
         ExecutorService executor = Executors.newCachedThreadPool();
         final CyclicBarrier barrier = new CyclicBarrier(Consumers + 1);
 
@@ -3465,7 +3468,7 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
 
                     } finally {
                         factory2.shutdown();
-                    }   
+                    }
                 });
 
         factory1.shutdown();

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -63,6 +63,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
+import lombok.Cleanup;
 import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper;
@@ -1271,6 +1272,7 @@ public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx)
     public void testConcurrentAsyncSetProperties() throws Exception {
         final CountDownLatch latch = new CountDownLatch(1000);
         ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));
+        @Cleanup("shutdownNow")
         ExecutorService executor = Executors.newCachedThreadPool();
         for (int i = 0; i < 1000; i++) {
             final int finalI = i;
@@ -1302,7 +1304,6 @@ public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx)
             fail(e.getMessage());
         }
         assertTrue(latch.await(300, TimeUnit.SECONDS));
-        executor.shutdown();
         factory.shutdown();
     }
 
@@ -1447,7 +1448,8 @@ public void testOpenRaceCondition() throws Exception {
 
         final int N = 1000;
         final Position position = ledger.addEntry("entry-0".getBytes());
-        Executor executor = Executors.newCachedThreadPool();
+        @Cleanup("shutdownNow")
+        ExecutorService executor = Executors.newCachedThreadPool();
         final CountDownLatch counter = new CountDownLatch(2);
         executor.execute(() -> {
             try {
@@ -2290,6 +2292,7 @@ public void testLazyRecoverCursor() throws Exception {
 
         // Simulating time consuming cursor recovery.
         CompletableFuture<Void> future = bkc.promiseAfter(2);
+        @Cleanup("shutdownNow")
         ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("lazyCursorRecovery"));
         scheduledExecutorService.schedule(() -> {
             future.complete(null);

managed-ledger/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java

@@ -126,7 +126,7 @@ public void tearDown() throws Exception {
         stopBKCluster();
         // stop zookeeper service
         stopZKCluster();
-        executor.shutdown();
+        executor.shutdownNow();
     }
 
     /**

managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java

@@ -119,10 +119,10 @@ public final void setUpClass() {
     @AfterClass(alwaysRun = true)
     public final void tearDownClass() {
         if (executor != null) {
-            executor.shutdown();
+            executor.shutdownNow();
         }
         if (cachedExecutor != null) {
-            cachedExecutor.shutdown();
+            cachedExecutor.shutdownNow();
         }
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/MessagingServiceShutdownHook.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
+import lombok.Cleanup;
 import org.apache.pulsar.zookeeper.ZooKeeperSessionWatcher.ShutdownService;
 import org.apache.zookeeper.ZooKeeper.States;
 import org.slf4j.ILoggerFactory;
@@ -52,6 +53,7 @@ public void run() {
                     + service.getSafeWebServiceAddress() + ", broker url=" + service.getSafeBrokerServiceUrl());
         }
 
+        @Cleanup("shutdownNow")
         ExecutorService executor = Executors.newSingleThreadExecutor(new DefaultThreadFactory("shutdown-thread"));
 
         try {

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -438,6 +438,11 @@ public CompletableFuture<Void> closeAsync() {
                 transactionBufferClient.close();
             }
 
+            if (transactionExecutor != null) {
+                transactionExecutor.shutdown();
+                transactionExecutor = null;
+            }
+
             if (coordinationService != null) {
                 coordinationService.close();
             }

pulsar-broker/src/test/java/org/apache/pulsar/broker/MockedBookKeeperClientFactory.java

@@ -74,6 +74,6 @@ public void close() {
             mockedBk.close();
         } catch (BKException | InterruptedException ignored) {
         }
-        executor.shutdown();
+        executor.shutdownNow();
     }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java

@@ -126,7 +126,7 @@ private void createTenant(PulsarAdmin pulsarAdmin)
     @AfterClass(alwaysRun = true)
     public void shutdown() throws Exception {
         log.info("--- Shutting down ---");
-        executor.shutdown();
+        executor.shutdownNow();
         executor = null;
 
         for (int i = 0; i < BROKER_COUNT; i++) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java

@@ -1346,6 +1346,7 @@ public void testNamespaceSplitBundleConcurrent() throws Exception {
             assertEquals(bundles.getBundles().get(i).toString(), splitRange[i]);
         }
 
+        @Cleanup("shutdownNow")
         ExecutorService executorService = Executors.newCachedThreadPool();
 
         try {
@@ -1403,7 +1404,6 @@ public void testNamespaceSplitBundleConcurrent() throws Exception {
         }
 
         producer.close();
-        executorService.shutdown();
     }
 
     @Test

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java

@@ -999,6 +999,7 @@ public void testNamespaceSplitBundleConcurrent() throws Exception {
             assertEquals(bundles.getBundles().get(i).toString(), splitRange[i]);
         }
 
+        @Cleanup("shutdownNow")
         ExecutorService executorService = Executors.newCachedThreadPool();
 
 
@@ -1083,7 +1084,6 @@ public void testNamespaceSplitBundleConcurrent() throws Exception {
         }
 
         producer.close();
-        executorService.shutdownNow();
     }
 
     @Test