[fix][test] Fix resource leaks in pulsar-metadata bookkeeper.replicat…

…ion tests (apache#21483)
streamnative · Oct 31, 2023 · b189ea7 · b189ea7
1 parent 69740c8
commit b189ea7
Show file tree

Hide file tree

Showing 7 changed files with 46 additions and 18 deletions.
diff --git a/pulsar-metadata/pom.xml b/pulsar-metadata/pom.xml
@@ -29,6 +29,11 @@
     <relativePath>..</relativePath>
   </parent>
 
+  <properties>
+    <!-- workaround for OOM error 134 in CI -->
+    <testForkCount>2</testForkCount>
+  </properties>
+
   <artifactId>pulsar-metadata</artifactId>
   <name>Pulsar Metadata</name>
   <dependencies>
@@ -149,7 +154,7 @@
           </execution>
         </executions>
       </plugin>
-      
+
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-jar-plugin</artifactId>

diff --git a/...adata/src/test/java/org/apache/bookkeeper/replication/AuditorCheckAllLedgersTaskTest.java b/...adata/src/test/java/org/apache/bookkeeper/replication/AuditorCheckAllLedgersTaskTest.java
@@ -65,15 +65,15 @@ public AuditorCheckAllLedgersTaskTest() {
     @Override
     public void setUp() throws Exception {
         super.setUp();
-        final BookKeeper bookKeeper = new BookKeeper(baseClientConf);
+        final BookKeeper bookKeeper = registerCloseable(new BookKeeper(baseClientConf));
         admin = new BookKeeperAdmin(bookKeeper, NullStatsLogger.INSTANCE, new ClientConfiguration(baseClientConf));
 
         String ledgersRoot = "/ledgers";
         String storeUri = metadataServiceUri.replaceAll("zk://", "").replaceAll("/ledgers", "");
-        MetadataStoreExtended store = MetadataStoreExtended.create(storeUri,
-                MetadataStoreConfig.builder().fsyncEnable(false).build());
+        MetadataStoreExtended store = registerCloseable(MetadataStoreExtended.create(storeUri,
+                MetadataStoreConfig.builder().fsyncEnable(false).build()));
         LayoutManager layoutManager = new PulsarLayoutManager(store, ledgersRoot);
-        PulsarLedgerManagerFactory ledgerManagerFactory = new PulsarLedgerManagerFactory();
+        PulsarLedgerManagerFactory ledgerManagerFactory = registerCloseable(new PulsarLedgerManagerFactory());
 
         ClientConfiguration conf = new ClientConfiguration();
         conf.setZkLedgersRootPath(ledgersRoot);
@@ -86,7 +86,7 @@ public void setUp() throws Exception {
                 acquireConcurrentOpenLedgerOperationsTimeoutMSec);
     }
 
-    @AfterMethod
+    @AfterMethod(alwaysRun = true)
     @Override
     public void tearDown() throws Exception {
         if (ledgerManager != null) {

diff --git a/...ar-metadata/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java b/...ar-metadata/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java
@@ -136,14 +136,14 @@ public void setUp() throws Exception {
 
         String ledgersRoot = "/ledgers";
         String storeUri = metadataServiceUri.replaceAll("zk://", "").replaceAll("/ledgers", "");
-        MetadataStoreExtended store = MetadataStoreExtended.create(storeUri,
-                MetadataStoreConfig.builder().fsyncEnable(false).build());
+        MetadataStoreExtended store = registerCloseable(MetadataStoreExtended.create(storeUri,
+                MetadataStoreConfig.builder().fsyncEnable(false).build()));
         LayoutManager layoutManager = new PulsarLayoutManager(store, ledgersRoot);
-        PulsarLedgerManagerFactory ledgerManagerFactory = new PulsarLedgerManagerFactory();
+        PulsarLedgerManagerFactory ledgerManagerFactory = registerCloseable(new PulsarLedgerManagerFactory());
         ClientConfiguration conf = new ClientConfiguration();
         conf.setZkLedgersRootPath(ledgersRoot);
         ledgerManagerFactory.initialize(conf, layoutManager, 1);
-        urLedgerMgr = ledgerManagerFactory.newLedgerUnderreplicationManager();
+        urLedgerMgr = registerCloseable(ledgerManagerFactory.newLedgerUnderreplicationManager());
         urLedgerMgr.setCheckAllLedgersCTime(System.currentTimeMillis());
 
         baseClientConf.setMetadataServiceUri(

diff --git a/.../src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTaskTest.java b/.../src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTaskTest.java
@@ -63,16 +63,19 @@ public void setUp() throws Exception {
         super.setUp();
         baseClientConf.setMetadataServiceUri(
                 metadataServiceUri.replaceAll("zk://", "metadata-store:").replaceAll("/ledgers", ""));
-        final BookKeeper bookKeeper = new BookKeeper(baseClientConf);
+        final BookKeeper bookKeeper = registerCloseable(new BookKeeper(baseClientConf));
         admin = new BookKeeperAdmin(bookKeeper, NullStatsLogger.INSTANCE, new ClientConfiguration(baseClientConf));
-        LedgerManagerFactory ledgerManagerFactory = bookKeeper.getLedgerManagerFactory();
+        LedgerManagerFactory ledgerManagerFactory = registerCloseable(bookKeeper.getLedgerManagerFactory());
         ledgerManager = ledgerManagerFactory.newLedgerManager();
         ledgerUnderreplicationManager = ledgerManagerFactory.newLedgerUnderreplicationManager();
     }
 
-    @AfterMethod
+    @AfterMethod(alwaysRun = true)
     @Override
     public void tearDown() throws Exception {
+        if (admin != null) {
+            admin.close();
+        }
         if (ledgerManager != null) {
             ledgerManager.close();
         }

diff --git a/...etadata/src/test/java/org/apache/bookkeeper/replication/AuditorReplicasCheckTaskTest.java b/...etadata/src/test/java/org/apache/bookkeeper/replication/AuditorReplicasCheckTaskTest.java
@@ -63,14 +63,14 @@ public void setUp() throws Exception {
         super.setUp();
         baseClientConf.setMetadataServiceUri(
                 metadataServiceUri.replaceAll("zk://", "metadata-store:").replaceAll("/ledgers", ""));
-        final BookKeeper bookKeeper = new BookKeeper(baseClientConf);
+        final BookKeeper bookKeeper = registerCloseable(new BookKeeper(baseClientConf));
         admin = new BookKeeperAdmin(bookKeeper, NullStatsLogger.INSTANCE, new ClientConfiguration(baseClientConf));
-        LedgerManagerFactory ledgerManagerFactory = bookKeeper.getLedgerManagerFactory();
+        LedgerManagerFactory ledgerManagerFactory = registerCloseable(bookKeeper.getLedgerManagerFactory());
         ledgerManager = ledgerManagerFactory.newLedgerManager();
         ledgerUnderreplicationManager = ledgerManagerFactory.newLedgerUnderreplicationManager();
     }
 
-    @AfterMethod
+    @AfterMethod(alwaysRun = true)
     @Override
     public void tearDown() throws Exception {
         if (ledgerManager != null) {

diff --git a/...r-metadata/src/test/java/org/apache/bookkeeper/replication/BookKeeperClusterTestCase.java b/...r-metadata/src/test/java/org/apache/bookkeeper/replication/BookKeeperClusterTestCase.java
@@ -115,6 +115,8 @@ public void handleTestMethodName(Method method) {
     protected ExecutorService executor;
     private final List<Integer> bookiePorts = new ArrayList<>();
 
+    private final List<AutoCloseable> closeables = new ArrayList<>();
+
     SynchronousQueue<Throwable> asyncExceptions = new SynchronousQueue<>();
     protected void captureThrowable(Runnable c) {
         try {
@@ -187,6 +189,9 @@ protected String changeLedgerPath() {
 
     @AfterTest(alwaysRun = true)
     public void tearDown() throws Exception {
+        callCloseables(closeables);
+        closeables.clear();
+
         boolean failed = false;
         for (Throwable e : asyncExceptions) {
             LOG.error("Got async exception: ", e);
@@ -228,6 +233,21 @@ public void tearDown() throws Exception {
         }
     }
 
+    protected <T extends AutoCloseable> T registerCloseable(T closeable) {
+        closeables.add(closeable);
+        return closeable;
+    }
+
+    private static void callCloseables(List<AutoCloseable> closeables) {
+        for (int i = closeables.size() - 1; i >= 0; i--) {
+            try {
+                closeables.get(i).close();
+            } catch (Exception e) {
+                LOG.error("Failure in calling close method", e);
+            }
+        }
+    }
+
     /**
      * Start zookeeper cluster.
      *

diff --git a/pulsar-metadata/src/test/java/org/apache/bookkeeper/replication/BookieLedgerIndexTest.java b/pulsar-metadata/src/test/java/org/apache/bookkeeper/replication/BookieLedgerIndexTest.java
@@ -91,8 +91,8 @@ public void setUp() throws Exception {
 
         String ledgersRoot = "/ledgers";
         String storeUri = metadataServiceUri.replaceAll("zk://", "").replaceAll("/ledgers", "");
-        MetadataStoreExtended store = MetadataStoreExtended.create(storeUri,
-                MetadataStoreConfig.builder().fsyncEnable(false).build());
+        MetadataStoreExtended store = registerCloseable(MetadataStoreExtended.create(storeUri,
+                MetadataStoreConfig.builder().fsyncEnable(false).build()));
         LayoutManager layoutManager = new PulsarLayoutManager(store, ledgersRoot);
         newLedgerManagerFactory = new PulsarLedgerManagerFactory();