[fix][test] Address flaky GetPartitionMetadataMultiBrokerTest (apache#23456)

(cherry picked from commit 0c4108b)

Author: lhotari

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

@@ -47,6 +47,7 @@
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.policies.data.TopicType;
 import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.tests.TestRetrySupport;
 import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
 import org.awaitility.Awaitility;
 import org.testng.annotations.AfterClass;
@@ -56,7 +57,7 @@
 
 @Test(groups = "broker-admin")
 @Slf4j
-public class GetPartitionMetadataTest {
+public class GetPartitionMetadataTest extends TestRetrySupport {
 
     protected static final String DEFAULT_NS = "public/default";
 
@@ -72,8 +73,10 @@ public class GetPartitionMetadataTest {
     protected PulsarClientImpl clientWithHttpLookup1;
     protected PulsarClientImpl clientWitBinaryLookup1;
 
+    @Override
     @BeforeClass(alwaysRun = true)
     protected void setup() throws Exception {
+        incrementSetupNumber();
         bkEnsemble = new LocalBookkeeperEnsemble(3, 0, () -> 0);
         bkEnsemble.start();
         // Start broker.
@@ -85,8 +88,10 @@ protected void setup() throws Exception {
         admin1.namespaces().createNamespace(DEFAULT_NS);
     }
 
+    @Override
     @AfterClass(alwaysRun = true)
     protected void cleanup() throws Exception {
+        markCurrentSetupNumberCleaned();
         cleanupBrokers();
         if (bkEnsemble != null) {
             bkEnsemble.stop();
@@ -236,7 +241,7 @@ public void testCompatibilityForNewClientAndOldBroker(TopicDomain topicDomain) t
         String pulsarUrl = pulsar1.getBrokerServiceUrl();
         PulsarClientImpl[] clients = getClientsToTest(false);
         for (PulsarClientImpl client : clients) {
-            client.getLookup(pulsarUrl).getBroker(TopicName.get(DEFAULT_NS + "/tp1"));
+            client.getLookup(pulsarUrl).getBroker(TopicName.get(DEFAULT_NS + "/tp1")).join();
         }
         // Inject a not support flag into the connections initialized.
         Field field = ClientCnx.class.getDeclaredField("supportsGetPartitionedMetadataWithoutAutoCreation");