[fix][test] Fix flaky testCreateTopicWithZombieReplicatorCursor (#20037)

(cherry picked from commit f2076b4)

Author: BewareMyPower

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

@@ -297,7 +297,7 @@ private void verifyActiveConsumerChange(CommandActiveConsumerChange change,
 
     @Test(timeOut = 10000)
     public void testAddConsumerWhenClosed() throws Exception {
-        PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService());
+        PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
         PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false);
         PersistentDispatcherSingleActiveConsumer pdfc = new PersistentDispatcherSingleActiveConsumer(cursorMock,
                 SubType.Failover, 0, topic, sub);

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java

@@ -34,7 +34,9 @@
 import static org.testng.Assert.assertNull;
 import com.google.common.collect.Sets;
 import java.lang.reflect.Field;
+import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -46,6 +48,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.client.LedgerHandle;
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedLedger;
@@ -72,6 +75,7 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+@Slf4j
 @Test(groups = "broker")
 public class PersistentTopicTest extends BrokerTestBase {
 
@@ -428,10 +432,10 @@ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy)
         admin.tenants().updateTenant("prop", tenantInfo);
 
         if (topicLevelPolicy) {
-            admin.topics().setReplicationClusters(topicName, Collections.singletonList(remoteCluster));
+            admin.topics().setReplicationClusters(topicName, Arrays.asList("test", remoteCluster));
         } else {
             admin.namespaces().setNamespaceReplicationClustersAsync(
-                    namespace, Collections.singleton(remoteCluster)).get();
+                    namespace, Sets.newHashSet("test", remoteCluster)).get();
         }
 
         final PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false)
@@ -446,16 +450,27 @@ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy)
         };
         assertEquals(getCursors.get(), Collections.singleton(conf.getReplicatorPrefix() + "." + remoteCluster));
 
+        // PersistentTopics#onPoliciesUpdate might happen in different threads, so there might be a race between two
+        // updates of the replication clusters. So here we sleep for a while to reduce the flakiness.
+        Thread.sleep(100);
+
+        // Configure the local cluster to avoid the topic being deleted in PersistentTopics#checkReplication
         if (topicLevelPolicy) {
-            admin.topics().setReplicationClusters(topicName, Collections.emptyList());
+            admin.topics().setReplicationClusters(topicName, Collections.singletonList("test"));
         } else {
-            admin.namespaces().setNamespaceReplicationClustersAsync(namespace, Collections.emptySet()).get();
+            admin.namespaces().setNamespaceReplicationClustersAsync(namespace, Collections.singleton("test")).get();
         }
         admin.clusters().deleteCluster(remoteCluster);
         // Now the cluster and its related policy has been removed but the replicator cursor still exists
 
-        topic.initialize().get(3, TimeUnit.SECONDS);
-        Awaitility.await().atMost(3, TimeUnit.SECONDS)
-                .until(() -> !topic.getManagedLedger().getCursors().iterator().hasNext());
+        Awaitility.await().atMost(Duration.ofSeconds(10)).until(() -> {
+            log.info("Before initialize...");
+            try {
+                topic.initialize().get(3, TimeUnit.SECONDS);
+            } catch (ExecutionException e) {
+                log.warn("Failed to initialize: {}", e.getCause().getMessage());
+            }
+            return !topic.getManagedLedger().getCursors().iterator().hasNext();
+        });
     }
 }