[fix][test] Fix thread resource leaks in tests and PrometheusMetricsProvider (#25487)

Author: lhotari

jetty-upgrade/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusMetricsProvider.java

@@ -21,6 +21,7 @@
 // CHECKSTYLE.OFF: IllegalImport
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.MoreExecutors;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import io.netty.util.internal.PlatformDependent;
 import io.prometheus.client.Collector;
@@ -171,6 +172,9 @@ public void stop() {
                 ThreadRegistry.clear();
             }
         }
+        if (executor != null) {
+            MoreExecutors.shutdownAndAwaitTermination(executor, 5, TimeUnit.SECONDS);
+        }
     }
 
     @Override

pulsar-broker-auth-athenz/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderAthenzTest.java

@@ -31,6 +31,7 @@
 import java.util.Properties;
 import javax.naming.AuthenticationException;
 import org.apache.pulsar.broker.ServiceConfiguration;
+import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
@@ -57,6 +58,13 @@ public void setup() throws Exception {
         System.setProperty(ZpeConsts.ZPE_PROP_ATHENZ_CONF, "./src/test/resources/athenz.conf.test");
     }
 
+    @AfterClass(alwaysRun = true)
+    public void cleanup() throws Exception {
+        if (provider != null) {
+            provider.close();
+        }
+    }
+
     @Test
     public void testInitilizeFromSystemPropeties() {
         System.setProperty("pulsar.athenz.domain.names", "test_provider");

pulsar-metadata/src/test/java/org/apache/bookkeeper/replication/BookKeeperClusterTestCase.java

@@ -28,6 +28,7 @@
 import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
 import static org.testng.Assert.assertFalse;
 import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.MoreExecutors;
 import java.io.File;
 import java.io.IOException;
 import java.lang.reflect.Method;
@@ -227,7 +228,7 @@ public void tearDown() throws Exception {
             tearDownException = e;
         }
 
-        executor.shutdownNow();
+        MoreExecutors.shutdownAndAwaitTermination(executor, 10, TimeUnit.SECONDS);
 
         LOG.info("Tearing down test {} in {} ms.", testName, sw.elapsed(TimeUnit.MILLISECONDS));
         if (tearDownException != null) {

pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/TransactionMetadataStoreProviderTest.java

@@ -34,6 +34,7 @@
 import org.apache.pulsar.transaction.coordinator.impl.TxnLogBufferedWriterConfig;
 import org.apache.pulsar.transaction.coordinator.proto.TxnStatus;
 import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Factory;
@@ -72,6 +73,13 @@ public void setup() throws Exception {
                 new TxnLogBufferedWriterConfig(), transactionTimer).get();
     }
 
+    @AfterMethod(alwaysRun = true)
+    public void teardown() throws Exception {
+        if (store != null) {
+            store.closeAsync().get(5, TimeUnit.SECONDS);
+        }
+    }
+
     @AfterClass
     public void cleanup(){
         transactionTimer.stop();