[fix][test] Fix resource leaks in ProxyTest and fix invalid tests (apache#24204)

(cherry picked from commit a3b31ff)
(cherry picked from commit bacd199)

Author: lhotari

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTest.java

@@ -154,12 +154,11 @@ protected void doInitConf() throws Exception {
     @Override
     @AfterClass(alwaysRun = true)
     protected void cleanup() throws Exception {
-        internalCleanup();
-
         proxyService.close();
         if (proxyClientAuthentication != null) {
             proxyClientAuthentication.close();
         }
+        internalCleanup();
     }
 
     @Test
@@ -245,6 +244,8 @@ public void testPartitions() throws Exception {
      **/
     @Test
     public void testAutoCreateTopic() throws Exception{
+        TopicType originalAllowAutoTopicCreationType = pulsar.getConfiguration().getAllowAutoTopicCreationType();
+        int originalDefaultNumPartitions = pulsar.getConfiguration().getDefaultNumPartitions();
         int defaultPartition = 2;
         int defaultNumPartitions = pulsar.getConfiguration().getDefaultNumPartitions();
         pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
@@ -256,10 +257,10 @@ public void testAutoCreateTopic() throws Exception{
             String topic = "persistent://sample/test/local/partitioned-proxy-topic";
             CompletableFuture<List<String>> partitionNamesFuture = client.getPartitionsForTopic(topic);
             List<String> partitionNames = partitionNamesFuture.get(30000, TimeUnit.MILLISECONDS);
-            Assert.assertEquals(partitionNames.size(), defaultPartition);
+            assertEquals(partitionNames.size(), defaultPartition);
         } finally {
-            pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
-            pulsar.getConfiguration().setDefaultNumPartitions(defaultNumPartitions);
+            pulsar.getConfiguration().setAllowAutoTopicCreationType(originalAllowAutoTopicCreationType);
+            pulsar.getConfiguration().setDefaultNumPartitions(originalDefaultNumPartitions);
         }
     }
 
@@ -318,8 +319,9 @@ public Object[][] topicTypes() {
         };
     }
 
-    @Test(timeOut = 60_000, dataProvider = "topicTypes", invocationCount = 100)
+    @Test(timeOut = 60_000, dataProvider = "topicTypes")
     public void testRegexSubscriptionWithTopicDiscovery(TopicType topicType) throws Exception {
+        @Cleanup
         final PulsarClient client = PulsarClient.builder().serviceUrl(proxyService.getServiceUrl()).build();
         final int topics = 10;
         final String subName = "s1";
@@ -394,7 +396,7 @@ public void testGetSchema() throws Exception {
         SchemaInfo schemaInfo = ((PulsarClientImpl) client).getLookup()
                 .getSchema(TopicName.get("persistent://sample/test/local/get-schema"), schemaVersion)
                 .get().orElse(null);
-        Assert.assertEquals(new String(schemaInfo.getSchema()), new String(schema.getSchemaInfo().getSchema()));
+        assertEquals(new String(schemaInfo.getSchema()), new String(schema.getSchemaInfo().getSchema()));
     }
 
     @Test
@@ -462,7 +464,7 @@ public void testGetClientVersion() throws Exception {
         PulsarClient client = PulsarClient.builder().serviceUrl(proxyService.getServiceUrl())
                 .build();
 
-        String topic = "persistent://sample/test/local/testGetClientVersion";
+        String topic = BrokerTestUtil.newUniqueName("persistent://sample/test/local/testGetClientVersion");
         String subName = "test-sub";
 
         @Cleanup
@@ -473,8 +475,8 @@ public void testGetClientVersion() throws Exception {
 
         consumer.receiveAsync();
 
-
-        Assert.assertEquals(admin.topics().getStats(topic).getSubscriptions().get(subName).getConsumers()
+        String partition = TopicName.get(topic).getPartition(0).toString();
+        assertEquals(admin.topics().getStats(partition).getSubscriptions().get(subName).getConsumers()
                 .get(0).getClientVersion(), String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()));
     }