[fix][tests] Fix resource leak in tests. Call cleanup before setup

- when setup is called explicitly, cleanup should be issued before it
- remove unnecessary "resetConfig" method calls
  - resetConfig gets called as part of internalCleanup
  - when resetConfig is in cleanup, it's possible to override config before calling setup
    - this allows using setup & cleanup methods instead of using
      error prone internalSetup and internalCleanup methods which might leave something behind

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

@@ -48,7 +48,6 @@ protected void setup() throws Exception {
     protected void cleanup() throws Exception {
         super.internalCleanup();
         useStaticPorts = false;
-        resetConfig();
     }
 
     @Override
@@ -105,6 +104,7 @@ public void testGetWorkerServiceException() throws Exception {
 
     @Test
     public void testAdvertisedAddress() throws Exception {
+        cleanup();
         useStaticPorts = true;
         setup();
         assertEquals(pulsar.getAdvertisedAddress(), "localhost");
@@ -117,6 +117,7 @@ public void testAdvertisedAddress() throws Exception {
 
     @Test
     public void testAdvertisedListeners() throws Exception {
+        cleanup();
         // don't use dynamic ports when using advertised listeners (#12079)
         useStaticPorts = true;
         conf.setAdvertisedListeners("internal:pulsar://gateway:6650, internal:pulsar+ssl://gateway:6651");
@@ -132,6 +133,7 @@ public void testAdvertisedListeners() throws Exception {
 
     @Test
     public void testDynamicBrokerPort() throws Exception {
+        cleanup();
         useStaticPorts = false;
         setup();
         assertEquals(pulsar.getAdvertisedAddress(), "localhost");

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

@@ -152,6 +152,7 @@ public void cleanup() throws Exception {
         super.internalCleanup();
         if (mockPulsarSetup != null) {
             mockPulsarSetup.cleanup();
+            mockPulsarSetup = null;
         }
         resetConfig();
     }
@@ -278,7 +279,7 @@ public void testIncrementPartitionsOfTopic() throws Exception {
     public void testTopicPoliciesWithMultiBroker() throws Exception {
         //setup cluster with 3 broker
         cleanup();
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test",
                 ClusterData.builder().serviceUrl((pulsar.getWebServiceAddress() + ",localhost:1026," + "localhost:2050")).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
@@ -655,19 +656,18 @@ public void testMaxConsumersOnSubApi() throws Exception {
     public void testLoadReportApi() throws Exception {
 
         this.conf.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
+        @Cleanup("cleanup")
         MockedPulsarService mockPulsarSetup1 = new MockedPulsarService(this.conf);
         mockPulsarSetup1.setup();
         PulsarAdmin simpleLoadManagerAdmin = mockPulsarSetup1.getAdmin();
         assertNotNull(simpleLoadManagerAdmin.brokerStats().getLoadReport());
 
         this.conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
+        @Cleanup("cleanup")
         MockedPulsarService mockPulsarSetup2 = new MockedPulsarService(this.conf);
         mockPulsarSetup2.setup();
         PulsarAdmin modularLoadManagerAdmin = mockPulsarSetup2.getAdmin();
         assertNotNull(modularLoadManagerAdmin.brokerStats().getLoadReport());
-
-        mockPulsarSetup1.cleanup();
-        mockPulsarSetup2.cleanup();
     }
 
     @Test
@@ -1442,7 +1442,7 @@ public void testDeleteTenant() throws Exception {
         // Disabled conf: systemTopicEnabled. see: https://github.com/apache/pulsar/pull/17070
         boolean originalSystemTopicEnabled = conf.isSystemTopicEnabled();
         if (originalSystemTopicEnabled) {
-            internalCleanup();
+            cleanup();
             conf.setSystemTopicEnabled(false);
             setup();
         }
@@ -1496,7 +1496,7 @@ public void testDeleteTenant() throws Exception {
         assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
         // Reset conf: systemTopicEnabled
         if (originalSystemTopicEnabled) {
-            internalCleanup();
+            cleanup();
             conf.setSystemTopicEnabled(true);
             setup();
         }
@@ -1535,7 +1535,7 @@ private void setNamespaceAttr(NamespaceAttr namespaceAttr){
     @Test(dataProvider = "namespaceAttrs")
     public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
         // Set conf.
-        internalCleanup();
+        cleanup();
         NamespaceAttr originalNamespaceAttr = markOriginalNamespaceAttr();
         setNamespaceAttr(namespaceAttr);
         setup();
@@ -1586,7 +1586,7 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
         assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
 
         // Reset config
-        internalCleanup();
+        cleanup();
         setNamespaceAttr(originalNamespaceAttr);
         setup();
     }
@@ -1636,7 +1636,7 @@ private void awaitChangeEventTopicAndCompactionCreateFinish(String ns, String to
 
     @Test
     public void testDeleteNamespaceWithTopicPolicies() throws Exception {
-        stopBroker();
+        cleanup();
         setup();
 
         String tenant = "test-tenant";
@@ -1940,9 +1940,9 @@ public void testUpdateClusterWithProxyUrl() throws Exception {
 
     @Test
     public void testMaxNamespacesPerTenant() throws Exception {
-        super.internalCleanup();
+        cleanup();
         conf.setMaxNamespacesPerTenant(2);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
         admin.tenants().createTenant("testTenant", tenantInfo);
@@ -1956,9 +1956,9 @@ public void testMaxNamespacesPerTenant() throws Exception {
         }
 
         //unlimited
-        super.internalCleanup();
+        cleanup();
         conf.setMaxNamespacesPerTenant(0);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         for (int i = 0; i < 10; i++) {
@@ -1968,9 +1968,9 @@ public void testMaxNamespacesPerTenant() throws Exception {
 
     @Test
     public void testAutoTopicCreationOverrideWithMaxNumPartitionsLimit() throws Exception{
-        super.internalCleanup();
+        cleanup();
         conf.setMaxNumPartitionsPerPartitionedTopic(10);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(
                 Set.of("role1", "role2"), Set.of("test"));
@@ -2010,9 +2010,9 @@ public void testAutoTopicCreationOverrideWithMaxNumPartitionsLimit() throws Exce
     }
     @Test
     public void testMaxTopicsPerNamespace() throws Exception {
-        super.internalCleanup();
+        cleanup();
         conf.setMaxTopicsPerNamespace(10);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
         admin.tenants().createTenant("testTenant", tenantInfo);
@@ -2033,9 +2033,9 @@ public void testMaxTopicsPerNamespace() throws Exception {
         }
 
         //unlimited
-        super.internalCleanup();
+        cleanup();
         conf.setMaxTopicsPerNamespace(0);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace("testTenant/ns1", Set.of("test"));
@@ -2045,9 +2045,9 @@ public void testMaxTopicsPerNamespace() throws Exception {
         }
 
         // check first create normal topic, then system topics, unlimited even setMaxTopicsPerNamespace
-        super.internalCleanup();
+        cleanup();
         conf.setMaxTopicsPerNamespace(5);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace("testTenant/ns1", Set.of("test"));
@@ -2058,9 +2058,9 @@ public void testMaxTopicsPerNamespace() throws Exception {
 
 
         // check first create system topics, then normal topic, unlimited even setMaxTopicsPerNamespace
-        super.internalCleanup();
+        cleanup();
         conf.setMaxTopicsPerNamespace(5);
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace("testTenant/ns1", Set.of("test"));
@@ -2070,11 +2070,11 @@ public void testMaxTopicsPerNamespace() throws Exception {
         }
 
         // check producer/consumer auto create partitioned topic
-        super.internalCleanup();
+        cleanup();
         conf.setMaxTopicsPerNamespace(10);
         conf.setDefaultNumPartitions(3);
         conf.setAllowAutoTopicCreationType("partitioned");
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace("testTenant/ns1", Set.of("test"));
@@ -2090,10 +2090,10 @@ public void testMaxTopicsPerNamespace() throws Exception {
         }
 
         // check producer/consumer auto create non-partitioned topic
-        super.internalCleanup();
+        cleanup();
         conf.setMaxTopicsPerNamespace(3);
         conf.setAllowAutoTopicCreationType("non-partitioned");
-        super.internalSetup();
+        setup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace("testTenant/ns1", Set.of("test"));
@@ -2125,9 +2125,9 @@ public void testInvalidBundleErrorResponse() throws Exception {
 
     @Test
     public void testMaxSubscriptionsPerTopic() throws Exception {
-        super.internalCleanup();
+        cleanup();
         conf.setMaxSubscriptionsPerTopic(2);
-        super.internalSetup();
+        setup();
 
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
@@ -2150,9 +2150,9 @@ public void testMaxSubscriptionsPerTopic() throws Exception {
             log.info("create subscription failed. Exception: ", e);
         }
 
-        super.internalCleanup();
+        cleanup();
         conf.setMaxSubscriptionsPerTopic(0);
-        super.internalSetup();
+        setup();
 
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
@@ -2166,9 +2166,9 @@ public void testMaxSubscriptionsPerTopic() throws Exception {
             admin.topics().createSubscription(topic, "test-sub" + i, MessageId.earliest);
         }
 
-        super.internalCleanup();
+        cleanup();
         conf.setMaxSubscriptionsPerTopic(2);
-        super.internalSetup();
+        setup();
 
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
@@ -2308,10 +2308,9 @@ public void testMaxSubPerTopic() throws Exception {
     @Test(timeOut = 30000)
     public void testMaxSubPerTopicPriority() throws Exception {
         final int brokerLevelMaxSub = 2;
-        super.internalCleanup();
-        mockPulsarSetup.cleanup();
+        cleanup();
         conf.setMaxSubscriptionsPerTopic(brokerLevelMaxSub);
-        super.internalSetup();
+        setup();
 
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
@@ -2365,10 +2364,9 @@ public void testMaxSubPerTopicPriority() throws Exception {
     @Test
     public void testMaxProducersPerTopicUnlimited() throws Exception {
         final int maxProducersPerTopic = 1;
-        super.internalCleanup();
-        mockPulsarSetup.cleanup();
+        cleanup();
         conf.setMaxProducersPerTopic(maxProducersPerTopic);
-        super.internalSetup();
+        setup();
         //init namespace
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
@@ -2418,10 +2416,9 @@ public void testMaxProducersPerTopicUnlimited() throws Exception {
     @Test
     public void testMaxConsumersPerTopicUnlimited() throws Exception {
         final int maxConsumersPerTopic = 1;
-        super.internalCleanup();
-        mockPulsarSetup.cleanup();
+        cleanup();
         conf.setMaxConsumersPerTopic(maxConsumersPerTopic);
-        super.internalSetup();
+        setup();
         //init namespace
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));

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

@@ -40,7 +40,6 @@ public class AdminApiClusterTest extends MockedPulsarServiceBaseTest {
     @BeforeMethod
     @Override
     public void setup() throws Exception {
-        resetConfig();
         super.internalSetup();
         admin.clusters()
                 .createCluster(CLUSTER, ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());

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

@@ -52,7 +52,6 @@ public class AdminApiHealthCheckTest extends MockedPulsarServiceBaseTest {
     @BeforeMethod
     @Override
     public void setup() throws Exception {
-        resetConfig();
         super.internalSetup();
         admin.clusters().createCluster("test",
                 ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());

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

@@ -59,7 +59,6 @@ public void setup() throws Exception {
     @Override
     public void cleanup() throws Exception {
         super.internalCleanup();
-        resetConfig();
     }
 
     @Test(timeOut = 30000)

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

@@ -103,7 +103,6 @@ private String parseResponseEntity(Object entity) throws Exception {
     @BeforeMethod
     @Override
     protected void setup() throws Exception {
-        resetConfig();
         super.internalSetup();
         // Create tenant, namespace, topic
         admin.clusters().createCluster(clusterName, ClusterData.builder().serviceUrl(brokerUrl.toString()).build());

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

@@ -67,7 +67,6 @@ protected void setup() throws Exception {
     @Override
     protected void cleanup() throws Exception {
         super.internalCleanup();
-        resetConfig();
     }
 
     @Test(timeOut = 10000)

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

@@ -159,7 +159,6 @@ public void initNamespace() throws Exception {
     @Override
     @BeforeMethod
     public void setup() throws Exception {
-        resetConfig();
         conf.setTopicLevelPoliciesEnabled(false);
         conf.setSystemTopicEnabled(false);
         conf.setClusterName(testLocalCluster);
@@ -1557,10 +1556,9 @@ public void testRetentionPolicyValidation() throws Exception {
     public void testMaxTopicsPerNamespace() throws Exception {
         cleanup();
         conf.setMaxTopicsPerNamespace(15);
-        super.internalSetup();
+        setup();
 
-        String namespace = "testTenant/ns1";
-        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+        String namespace = BrokerTestUtil.newUniqueName("testTenant/ns1");
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"),
                 Set.of("use"));
         admin.tenants().createTenant("testTenant", tenantInfo);
@@ -1572,7 +1570,7 @@ public void testMaxTopicsPerNamespace() throws Exception {
         assertEquals(10, admin.namespaces().getMaxTopicsPerNamespace(namespace));
 
         // check create partitioned/non-partitioned topics using namespace policy
-        String topic = "persistent://testTenant/ns1/test_create_topic_v";
+        String topic = "persistent://" + namespace + "/test_create_topic_v";
         admin.topics().createPartitionedTopic(topic + "1", 2);
         admin.topics().createPartitionedTopic(topic + "2", 3);
         admin.topics().createPartitionedTopic(topic + "3", 4);
@@ -1610,9 +1608,8 @@ public void testMaxTopicsPerNamespace() throws Exception {
         conf.setMaxTopicsPerNamespace(0);
         conf.setDefaultNumPartitions(3);
         conf.setAllowAutoTopicCreationType("partitioned");
-        super.internalSetup();
+        setup();
 
-        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace(namespace, Set.of("use"));
         admin.namespaces().setMaxTopicsPerNamespace(namespace, 10);
@@ -1640,9 +1637,8 @@ public void testMaxTopicsPerNamespace() throws Exception {
         conf.setMaxTopicsPerNamespace(0);
         conf.setDefaultNumPartitions(1);
         conf.setAllowAutoTopicCreationType("non-partitioned");
-        super.internalSetup();
+        setup();
 
-        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace(namespace, Set.of("use"));
         admin.namespaces().setMaxTopicsPerNamespace(namespace, 3);