[fix] [test] Fix flaky test ReplicatorTest (apache#22594)

poorbarcode · poorbarcode · commit 9242f333d9ce · 2024-05-07T20:00:21.000+08:00
(cherry picked from commit 6fdc0e3)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
@@ -18,15 +18,22 @@
  */
 package org.apache.pulsar.broker.service;
 
+import static org.testng.Assert.fail;
 import com.google.common.collect.Sets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
 import lombok.Cleanup;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.impl.ConsumerImpl;
 import org.apache.pulsar.client.impl.ProducerImpl;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -36,6 +43,11 @@
 import java.lang.reflect.Method;
 import java.util.concurrent.TimeUnit;
 
+/**
+ * The tests in this class should be denied in a production pulsar cluster. they are very dangerous, which leads to
+ * a lot of topic deletion and makes namespace policies being incorrect.
+ */
+@Slf4j
 @Test(groups = "broker-impl")
 public class ReplicatorGlobalNSTest extends ReplicatorTestBase {
 
@@ -64,7 +76,7 @@ public void cleanup() throws Exception {
      *
      * @throws Exception
      */
-    @Test
+    @Test(priority = Integer.MAX_VALUE)
     public void testRemoveLocalClusterOnGlobalNamespace() throws Exception {
         log.info("--- Starting ReplicatorTest::testRemoveLocalClusterOnGlobalNamespace ---");
 
@@ -100,32 +112,88 @@ public void testRemoveLocalClusterOnGlobalNamespace() throws Exception {
 
     }
 
-    @Test
-    public void testForcefullyTopicDeletion() throws Exception {
-        log.info("--- Starting ReplicatorTest::testForcefullyTopicDeletion ---");
-
-        final String namespace = "pulsar/removeClusterTest";
-        admin1.namespaces().createNamespace(namespace);
-        admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1"));
-
-        final String topicName = "persistent://" + namespace + "/topic";
-
-        @Cleanup
-        PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS)
-                .build();
-
-        ProducerImpl<byte[]> producer1 = (ProducerImpl<byte[]>) client1.newProducer().topic(topicName)
-                .enableBatching(false).messageRoutingMode(MessageRoutingMode.SinglePartition).create();
-        producer1.close();
-
-        admin1.topics().delete(topicName, true);
-
-        MockedPulsarServiceBaseTest
-                .retryStrategically((test) -> !pulsar1.getBrokerService().getTopics().containsKey(topicName), 50, 150);
-
-        Assert.assertFalse(pulsar1.getBrokerService().getTopics().containsKey(topicName));
+    /**
+     * This is not a formal operation and can cause serious problems if call it in a production environment.
+     */
+    @Test(priority = Integer.MAX_VALUE - 1)
+    public void testConfigChange() throws Exception {
+        log.info("--- Starting ReplicatorTest::testConfigChange ---");
+        // This test is to verify that the config change on global namespace is successfully applied in broker during
+        // runtime.
+        // Run a set of producer tasks to create the topics
+        List<Future<Void>> results = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            final TopicName dest = TopicName.get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/topic-" + i));
+
+            results.add(executor.submit(new Callable<Void>() {
+                @Override
+                public Void call() throws Exception {
+
+                    @Cleanup
+                    MessageProducer producer = new MessageProducer(url1, dest);
+                    log.info("--- Starting producer --- " + url1);
+
+                    @Cleanup
+                    MessageConsumer consumer = new MessageConsumer(url1, dest);
+                    log.info("--- Starting Consumer --- " + url1);
+
+                    producer.produce(2);
+                    consumer.receive(2);
+                    return null;
+                }
+            }));
+        }
+
+        for (Future<Void> result : results) {
+            try {
+                result.get();
+            } catch (Exception e) {
+                log.error("exception in getting future result ", e);
+                fail(String.format("replication test failed with %s exception", e.getMessage()));
+            }
+        }
+
+        Thread.sleep(1000L);
+        // Make sure that the internal replicators map contains remote cluster info
+        ConcurrentOpenHashMap<String, PulsarClient> replicationClients1 = ns1.getReplicationClients();
+        ConcurrentOpenHashMap<String, PulsarClient> replicationClients2 = ns2.getReplicationClients();
+        ConcurrentOpenHashMap<String, PulsarClient> replicationClients3 = ns3.getReplicationClients();
+
+        Assert.assertNotNull(replicationClients1.get("r2"));
+        Assert.assertNotNull(replicationClients1.get("r3"));
+        Assert.assertNotNull(replicationClients2.get("r1"));
+        Assert.assertNotNull(replicationClients2.get("r3"));
+        Assert.assertNotNull(replicationClients3.get("r1"));
+        Assert.assertNotNull(replicationClients3.get("r2"));
+
+        // Case 1: Update the global namespace replication configuration to only contains the local cluster itself
+        admin1.namespaces().setNamespaceReplicationClusters("pulsar/ns", Sets.newHashSet("r1"));
+
+        // Wait for config changes to be updated.
+        Thread.sleep(1000L);
+
+        // Make sure that the internal replicators map still contains remote cluster info
+        Assert.assertNotNull(replicationClients1.get("r2"));
+        Assert.assertNotNull(replicationClients1.get("r3"));
+        Assert.assertNotNull(replicationClients2.get("r1"));
+        Assert.assertNotNull(replicationClients2.get("r3"));
+        Assert.assertNotNull(replicationClients3.get("r1"));
+        Assert.assertNotNull(replicationClients3.get("r2"));
+
+        // Case 2: Update the configuration back
+        admin1.namespaces().setNamespaceReplicationClusters("pulsar/ns", Sets.newHashSet("r1", "r2", "r3"));
+
+        // Wait for config changes to be updated.
+        Thread.sleep(1000L);
+
+        // Make sure that the internal replicators map still contains remote cluster info
+        Assert.assertNotNull(replicationClients1.get("r2"));
+        Assert.assertNotNull(replicationClients1.get("r3"));
+        Assert.assertNotNull(replicationClients2.get("r1"));
+        Assert.assertNotNull(replicationClients2.get("r3"));
+        Assert.assertNotNull(replicationClients3.get("r1"));
+        Assert.assertNotNull(replicationClients3.get("r2"));
+
+        // Case 3: TODO: Once automatic cleanup is implemented, add tests case to verify auto removal of clusters
     }
-
-    private static final Logger log = LoggerFactory.getLogger(ReplicatorGlobalNSTest.class);
-
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -44,13 +44,11 @@
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.UUID;
-import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
@@ -68,6 +66,7 @@
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.service.BrokerServiceException.NamingException;
 import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -154,88 +153,6 @@ public Object[][] partitionedTopicProvider() {
         return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
     }
 
-    @Test(priority = Integer.MAX_VALUE)
-    public void testConfigChange() throws Exception {
-        log.info("--- Starting ReplicatorTest::testConfigChange ---");
-        // This test is to verify that the config change on global namespace is successfully applied in broker during
-        // runtime.
-        // Run a set of producer tasks to create the topics
-        List<Future<Void>> results = new ArrayList<>();
-        for (int i = 0; i < 10; i++) {
-            final TopicName dest = TopicName.get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/topic-" + i));
-
-            results.add(executor.submit(new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-
-                    @Cleanup
-                    MessageProducer producer = new MessageProducer(url1, dest);
-                    log.info("--- Starting producer --- " + url1);
-
-                    @Cleanup
-                    MessageConsumer consumer = new MessageConsumer(url1, dest);
-                    log.info("--- Starting Consumer --- " + url1);
-
-                    producer.produce(2);
-                    consumer.receive(2);
-                    return null;
-                }
-            }));
-        }
-
-        for (Future<Void> result : results) {
-            try {
-                result.get();
-            } catch (Exception e) {
-                log.error("exception in getting future result ", e);
-                fail(String.format("replication test failed with %s exception", e.getMessage()));
-            }
-        }
-
-        Thread.sleep(1000L);
-        // Make sure that the internal replicators map contains remote cluster info
-        ConcurrentOpenHashMap<String, PulsarClient> replicationClients1 = ns1.getReplicationClients();
-        ConcurrentOpenHashMap<String, PulsarClient> replicationClients2 = ns2.getReplicationClients();
-        ConcurrentOpenHashMap<String, PulsarClient> replicationClients3 = ns3.getReplicationClients();
-
-        Assert.assertNotNull(replicationClients1.get("r2"));
-        Assert.assertNotNull(replicationClients1.get("r3"));
-        Assert.assertNotNull(replicationClients2.get("r1"));
-        Assert.assertNotNull(replicationClients2.get("r3"));
-        Assert.assertNotNull(replicationClients3.get("r1"));
-        Assert.assertNotNull(replicationClients3.get("r2"));
-
-        // Case 1: Update the global namespace replication configuration to only contains the local cluster itself
-        admin1.namespaces().setNamespaceReplicationClusters("pulsar/ns", Sets.newHashSet("r1"));
-
-        // Wait for config changes to be updated.
-        Thread.sleep(1000L);
-
-        // Make sure that the internal replicators map still contains remote cluster info
-        Assert.assertNotNull(replicationClients1.get("r2"));
-        Assert.assertNotNull(replicationClients1.get("r3"));
-        Assert.assertNotNull(replicationClients2.get("r1"));
-        Assert.assertNotNull(replicationClients2.get("r3"));
-        Assert.assertNotNull(replicationClients3.get("r1"));
-        Assert.assertNotNull(replicationClients3.get("r2"));
-
-        // Case 2: Update the configuration back
-        admin1.namespaces().setNamespaceReplicationClusters("pulsar/ns", Sets.newHashSet("r1", "r2", "r3"));
-
-        // Wait for config changes to be updated.
-        Thread.sleep(1000L);
-
-        // Make sure that the internal replicators map still contains remote cluster info
-        Assert.assertNotNull(replicationClients1.get("r2"));
-        Assert.assertNotNull(replicationClients1.get("r3"));
-        Assert.assertNotNull(replicationClients2.get("r1"));
-        Assert.assertNotNull(replicationClients2.get("r3"));
-        Assert.assertNotNull(replicationClients3.get("r1"));
-        Assert.assertNotNull(replicationClients3.get("r2"));
-
-        // Case 3: TODO: Once automatic cleanup is implemented, add tests case to verify auto removal of clusters
-    }
-
     @Test(timeOut = 10000)
     public void activeBrokerParse() throws Exception {
         pulsar1.getConfiguration().setAuthorizationEnabled(true);
@@ -253,6 +170,32 @@ public void activeBrokerParse() throws Exception {
         pulsar1.getConfiguration().setAuthorizationEnabled(false);
     }
 
+    @Test
+    public void testForcefullyTopicDeletion() throws Exception {
+        log.info("--- Starting ReplicatorTest::testForcefullyTopicDeletion ---");
+
+        final String namespace = BrokerTestUtil.newUniqueName("pulsar/removeClusterTest");
+        admin1.namespaces().createNamespace(namespace);
+        admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1"));
+
+        final String topicName = "persistent://" + namespace + "/topic";
+
+        @Cleanup
+        PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS)
+                .build();
+
+        ProducerImpl<byte[]> producer1 = (ProducerImpl<byte[]>) client1.newProducer().topic(topicName)
+                .enableBatching(false).messageRoutingMode(MessageRoutingMode.SinglePartition).create();
+        producer1.close();
+
+        admin1.topics().delete(topicName, true);
+
+        MockedPulsarServiceBaseTest
+                .retryStrategically((test) -> !pulsar1.getBrokerService().getTopics().containsKey(topicName), 50, 150);
+
+        Assert.assertFalse(pulsar1.getBrokerService().getTopics().containsKey(topicName));
+    }
+
     @SuppressWarnings("unchecked")
     @Test(timeOut = 30000)
     public void testConcurrentReplicator() throws Exception {
@@ -1240,7 +1183,7 @@ public void testReplicatedCluster() throws Exception {
 
         log.info("--- Starting ReplicatorTest::testReplicatedCluster ---");
 
-        final String namespace = "pulsar/global/repl";
+        final String namespace = BrokerTestUtil.newUniqueName("pulsar/global/repl");
         final String topicName = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/topic1");
         admin1.namespaces().createNamespace(namespace);
         admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3"));
@@ -1647,7 +1590,7 @@ public void testReplicatorWithFailedAck() throws Exception {
 
         log.info("--- Starting ReplicatorTest::testReplication ---");
 
-        String namespace = "pulsar/global/ns2";
+        String namespace = BrokerTestUtil.newUniqueName("pulsar/global/ns");
         admin1.namespaces().createNamespace(namespace, Sets.newHashSet("r1"));
         final TopicName dest = TopicName
                 .get(BrokerTestUtil.newUniqueName("persistent://" + namespace + "/ackFailedTopic"));
@@ -1685,13 +1628,15 @@ public void testReplicatorWithFailedAck() throws Exception {
 
         MessageIdImpl lastMessageId = (MessageIdImpl) topic.getLastMessageId().get();
         Position lastPosition = PositionImpl.get(lastMessageId.getLedgerId(), lastMessageId.getEntryId());
-        ConcurrentOpenHashMap<String, Replicator> replicators = topic.getReplicators();
-        PersistentReplicator replicator = (PersistentReplicator) replicators.get("r2");
 
         Awaitility.await().pollInterval(1, TimeUnit.SECONDS).timeout(30, TimeUnit.SECONDS)
-                .untilAsserted(() -> assertEquals(org.apache.pulsar.broker.service.AbstractReplicator.State.Started,
-                        replicator.getState()));
-        assertEquals(replicator.getState(), org.apache.pulsar.broker.service.AbstractReplicator.State.Started);
+                .ignoreExceptions()
+                .untilAsserted(() -> {
+                    ConcurrentOpenHashMap<String, Replicator> replicators = topic.getReplicators();
+                    PersistentReplicator replicator = (PersistentReplicator) replicators.get("r2");
+                    assertEquals(org.apache.pulsar.broker.service.AbstractReplicator.State.Started,
+                            replicator.getState());
+                });
 
         // Make sure all the data has replicated to the remote cluster before close the cursor.
         Awaitility.await().untilAsserted(() -> assertEquals(cursor.getMarkDeletedPosition(), lastPosition));
@@ -1717,7 +1662,7 @@ public void testReplicatorWithFailedAck() throws Exception {
     @Test
     public void testWhenUpdateReplicationCluster() throws Exception {
         log.info("--- testWhenUpdateReplicationCluster ---");
-        String namespace = "pulsar/ns2";
+        String namespace = BrokerTestUtil.newUniqueName("pulsar/ns");;
         admin1.namespaces().createNamespace(namespace);
         admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2"));
         final TopicName dest = TopicName.get(
@@ -1746,12 +1691,12 @@ public void testWhenUpdateReplicationCluster() throws Exception {
     @Test
     public void testReplicatorProducerNotExceed() throws Exception {
         log.info("--- testReplicatorProducerNotExceed ---");
-        String namespace1 = "pulsar/ns11";
+        String namespace1 = BrokerTestUtil.newUniqueName("pulsar/ns1");
         admin1.namespaces().createNamespace(namespace1);
         admin1.namespaces().setNamespaceReplicationClusters(namespace1, Sets.newHashSet("r1", "r2"));
         final TopicName dest1 = TopicName.get(
                 BrokerTestUtil.newUniqueName("persistent://" + namespace1 + "/testReplicatorProducerNotExceed1"));
-        String namespace2 = "pulsar/ns22";
+        String namespace2 = BrokerTestUtil.newUniqueName("pulsar/ns2");
         admin2.namespaces().createNamespace(namespace2);
         admin2.namespaces().setNamespaceReplicationClusters(namespace2, Sets.newHashSet("r1", "r2"));
         final TopicName dest2 = TopicName.get(
