[fix][test] Clear fields in test cleanup to reduce memory consumption

- TestNG keeps a reference to the test instance and this will cause
  a OOME in some cases

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

@@ -59,6 +59,7 @@ protected void onCleanup() {
             } catch (Exception e) {
                 log.error("Error in stopping ZK server", e);
             }
+            testZKServer = null;
         }
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CanReconnectZKClientPulsarServiceBaseTest.java

@@ -43,7 +43,6 @@
 
 @Slf4j
 public abstract class CanReconnectZKClientPulsarServiceBaseTest extends TestRetrySupport {
-
     protected final String defaultTenant = "public";
     protected final String defaultNamespace = defaultTenant + "/default";
     protected int numberOfBookies = 3;
@@ -60,6 +59,7 @@ public abstract class CanReconnectZKClientPulsarServiceBaseTest extends TestRetr
     protected ZooKeeper localZkOfBroker;
     protected Object localMetaDataStoreClientCnx;
     protected final AtomicBoolean LocalMetadataStoreInReconnectFinishSignal = new AtomicBoolean();
+
     protected void startZKAndBK() throws Exception {
         // Start ZK.
         brokerConfigZk = new ZookeeperServerTest(0);
@@ -198,15 +198,28 @@ protected void cleanup() throws Exception {
         stopLocalMetadataStoreAlwaysReconnect();
 
         // Stop brokers.
-        client.close();
-        admin.close();
+        if (client != null) {
+            client.close();
+            client = null;
+        }
+        if (admin != null) {
+            admin.close();
+            admin = null;
+        }
         if (pulsar != null) {
             pulsar.close();
+            pulsar = null;
         }
 
         // Stop ZK and BK.
-        bkEnsemble.stop();
-        brokerConfigZk.stop();
+        if (bkEnsemble != null) {
+            bkEnsemble.stop();
+            bkEnsemble = null;
+        }
+        if (brokerConfigZk != null) {
+            brokerConfigZk.stop();
+            brokerConfigZk = null;
+        }
 
         // Reset configs.
         config = new ServiceConfiguration();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTestBase.java

@@ -386,29 +386,60 @@ protected void cleanup() throws Exception {
             executor = null;
         }
 
-        admin1.close();
-        admin2.close();
-        admin3.close();
-        admin4.close();
+        if (admin1 != null) {
+            admin1.close();
+            admin1 = null;
+        }
+        if (admin2 != null) {
+            admin2.close();
+            admin2 = null;
+        }
+        if (admin3 != null) {
+            admin3.close();
+            admin3 = null;
+        }
+        if (admin4 != null) {
+            admin4.close();
+            admin4 = null;
+        }
 
         if (pulsar4 != null) {
             pulsar4.close();
+            pulsar4 = null;
         }
         if (pulsar3 != null) {
             pulsar3.close();
+            pulsar3 = null;
         }
         if (pulsar2 != null) {
             pulsar2.close();
+            pulsar2 = null;
         }
         if (pulsar1 != null) {
             pulsar1.close();
+            pulsar1 = null;
         }
 
-        bkEnsemble1.stop();
-        bkEnsemble2.stop();
-        bkEnsemble3.stop();
-        bkEnsemble4.stop();
-        globalZkS.stop();
+        if (bkEnsemble1 != null) {
+            bkEnsemble1.stop();
+            bkEnsemble1 = null;
+        }
+        if (bkEnsemble2 != null) {
+            bkEnsemble2.stop();
+            bkEnsemble2 = null;
+        }
+        if (bkEnsemble3 != null) {
+            bkEnsemble3.stop();
+            bkEnsemble3 = null;
+        }
+        if (bkEnsemble4 != null) {
+            bkEnsemble4.stop();
+            bkEnsemble4 = null;
+        }
+        if (globalZkS != null) {
+            globalZkS.stop();
+            globalZkS = null;
+        }
 
         resetConfig1();
         resetConfig2();