[fix][test] Fix more resource leaks in tests (#24314)

(cherry picked from commit eccc6b6)

Author: lhotari

buildtools/src/main/java/org/apache/pulsar/tests/ThreadLeakDetectorListener.java

@@ -227,6 +227,10 @@ private static boolean shouldSkipThread(Thread thread) {
             if (threadName.equals("process reaper")) {
                 return true;
             }
+            // skip thread created by sun.net.www.http.KeepAliveCache
+            if (threadName.equals("Keep-Alive-Timer")) {
+                return true;
+            }
             // skip JVM internal thread related to agent attach
             if (threadName.equals("Attach Listener")) {
                 return true;
@@ -255,6 +259,10 @@ private static boolean shouldSkipThread(Thread thread) {
             if (threadName.equals("Grizzly-HttpSession-Expirer")) {
                 return true;
             }
+            // skip Hadoop LocalFileSystem stats thread
+            if (threadName.equals("org.apache.hadoop.fs.FileSystem$Statistics$StatisticsDataReferenceCleaner")) {
+                return true;
+            }
             // Testcontainers AbstractWaitStrategy.EXECUTOR
             if (threadName.startsWith("testcontainers-wait-")) {
                 return true;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -250,7 +250,7 @@ private ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore,
         openTelemetryManagedCursorStats = new OpenTelemetryManagedCursorStats(openTelemetry, this);
     }
 
-    static class DefaultBkFactory implements BookkeeperFactoryForCustomEnsemblePlacementPolicy {
+    static class DefaultBkFactory implements BookkeeperFactoryForCustomEnsemblePlacementPolicy, AutoCloseable {
 
         private final BookKeeper bkClient;
 
@@ -263,6 +263,11 @@ public DefaultBkFactory(ClientConfiguration bkClientConfiguration)
         public CompletableFuture<BookKeeper> get(EnsemblePlacementPolicyConfig policy) {
             return CompletableFuture.completedFuture(bkClient);
         }
+
+        @Override
+        public void close() throws Exception {
+            bkClient.close();
+        }
     }
 
     private synchronized void handleMetadataStoreNotification(SessionEvent e) {
@@ -646,13 +651,20 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
                             }
                         }
                     }));
-                }).thenAcceptAsync(__ -> {
+                }).whenCompleteAsync((__, ___) -> {
                     //wait for tasks in scheduledExecutor executed.
                     openTelemetryManagedCursorStats.close();
                     openTelemetryManagedLedgerStats.close();
                     openTelemetryCacheStats.close();
                     scheduledExecutor.shutdownNow();
                     entryCacheManager.clear();
+                    if (bookkeeperFactory instanceof DefaultBkFactory defaultBkFactory) {
+                        try {
+                            defaultBkFactory.close();
+                        } catch (Exception e) {
+                            log.warn("Failed to close bookkeeper client", e);
+                        }
+                    }
                 });
     }
 

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

@@ -26,7 +26,6 @@
 import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
 import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
 import static org.testng.Assert.assertFalse;
-
 import com.google.common.base.Stopwatch;
 import java.io.File;
 import java.io.IOException;
@@ -73,9 +72,9 @@
 import org.awaitility.Awaitility;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.testng.annotations.AfterTest;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.BeforeTest;
 
 /**
  * A class runs several bookie servers for testing.
@@ -148,7 +147,7 @@ public BookKeeperClusterTestCase(int numBookies, int numOfZKNodes, int testTimeo
         }
     }
 
-    @BeforeTest(alwaysRun = true)
+    @BeforeClass(alwaysRun = true)
     public void setUp() throws Exception {
         setUp(getLedgersRootPath());
     }
@@ -187,7 +186,7 @@ protected String changeLedgerPath() {
         return "";
     }
 
-    @AfterTest(alwaysRun = true)
+    @AfterClass(alwaysRun = true)
     public void tearDown() throws Exception {
         boolean failed = false;
         for (Throwable e : asyncExceptions) {