[FlakyTest]Try to fix flaky test ReplicatorTest.testReplicatorOnParti…

…tionedTopic. (apache#9659)

Fixes apache#9457

### Motivation
Not sure why state is not cleaned up but exception is indicating metadata of namespace the test trying to create already exist so use a different namespace name for different test run should able to fix it.

### Modifications
Append System.nanoTime to the namespace used to avoid being affect by previous state. This seems working fine for other test cases in the same test.

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

@@ -0,0 +1,32 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker;
+
+import java.util.UUID;
+
+/**
+ * Holds util methods used in test.
+ */
+public class BrokerTestUtil {
+    // Generate unique name for different test run.
+    public static String newUniqueName(String prefix) {
+        return prefix + "-" + UUID.randomUUID();
+    }
+
+}

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

@@ -60,6 +60,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.ConfigHelper;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.PulsarService;
@@ -2621,7 +2622,7 @@ public void testCreateAndDeleteNamespaceWithBundles() throws Exception {
                 Sets.newHashSet("test", "usw"));
         admin.tenants().updateTenant("prop-xyz", tenantInfo);
 
-        String ns = "prop-xyz/ns-" + System.nanoTime();
+        String ns = BrokerTestUtil.newUniqueName("prop-xyz/ns");
 
         admin.namespaces().createNamespace(ns, 24);
         admin.namespaces().deleteNamespace(ns);

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.TimeUnit;
 
 import lombok.Cleanup;
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.client.api.Consumer;
@@ -107,7 +108,8 @@ public void testIncrementPartitionsOfTopic() throws Exception {
 
     @Test
     public void testIncrementPartitionsWithNoSubscriptions() throws Exception {
-        final String partitionedTopicName = "persistent://prop-xyz/use/ns1/test-topic-" + System.nanoTime();
+        final String partitionedTopicName =
+                BrokerTestUtil.newUniqueName("persistent://prop-xyz/use/ns1/test-topic");
 
         admin.topics().createPartitionedTopic(partitionedTopicName, 1);
         assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 1);
@@ -133,7 +135,8 @@ public void testIncrementPartitionsWithNoSubscriptions() throws Exception {
 
     @Test
     public void testIncrementPartitionsWithReaders() throws Exception {
-        TopicName partitionedTopicName = TopicName.get("persistent://prop-xyz/use/ns1/test-topic-" + System.nanoTime());
+        TopicName partitionedTopicName = TopicName.get(
+                BrokerTestUtil.newUniqueName("persistent://prop-xyz/use/ns1/test-topic"));
 
         admin.topics().createPartitionedTopic(partitionedTopicName.toString(), 1);
         assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName.toString()).partitions, 1);

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

@@ -20,6 +20,7 @@
 
 import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
 
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.PulsarClientException;
 import static org.mockito.ArgumentMatchers.any;
@@ -1140,7 +1141,7 @@ public void testDeleteNamespace() throws Exception {
      */
     @Test
     public void testForceDeleteNamespace() throws Exception {
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
         String topic = namespace + "/topic";
 
         admin.namespaces().createNamespace(namespace, 100);
@@ -1360,7 +1361,7 @@ private void mockWebUrl(URL localWebServiceUrl, NamespaceName namespace) throws
 
     @Test
     public void testDeleteNonPartitionedTopicMultipleTimes() throws Exception {
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
         String topic = namespace + "/topic";
 
         admin.namespaces().createNamespace(namespace, Sets.newHashSet(testLocalCluster));
@@ -1388,7 +1389,7 @@ public void testDeleteNonPartitionedTopicMultipleTimes() throws Exception {
 
     @Test
     public void testDeletePartitionedTopicMultipleTimes() throws Exception {
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
         String topic = namespace + "/topic";
 
         admin.namespaces().createNamespace(namespace, Sets.newHashSet(testLocalCluster));
@@ -1417,7 +1418,7 @@ public void testDeletePartitionedTopicMultipleTimes() throws Exception {
 
     @Test
     public void testRetentionPolicyValidation() throws Exception {
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
 
         admin.namespaces().createNamespace(namespace, Sets.newHashSet(testLocalCluster));
 
@@ -1596,7 +1597,7 @@ public void testRetentionPolicyValidationAsPartOfAllPolicies() throws Exception
     public void testSubscriptionTypesEnabled() throws PulsarAdminException, PulsarClientException {
         pulsar.getConfiguration().setAuthorizationEnabled(false);
         pulsar.getConfiguration().setTopicLevelPoliciesEnabled(false);
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
         String topic = namespace + "/test-subscription-enabled";
         admin.namespaces().createNamespace(namespace);
         Set<SubscriptionType> subscriptionTypes = new HashSet<>();
@@ -1666,7 +1667,7 @@ public void testSubscriptionTypesEnabled() throws PulsarAdminException, PulsarCl
 
     private void assertValidRetentionPolicyAsPartOfAllPolicies(Policies policies, int retentionTimeInMinutes,
                                                                int retentionSizeInMB) throws PulsarAdminException {
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
         RetentionPolicies retention = new RetentionPolicies(retentionTimeInMinutes, retentionSizeInMB);
         policies.retention_policies = retention;
         admin.namespaces().createNamespace(namespace, policies);
@@ -1675,7 +1676,7 @@ private void assertValidRetentionPolicyAsPartOfAllPolicies(Policies policies, in
 
     private void assertInvalidRetentionPolicyAsPartOfAllPolicies(Policies policies, int retentionTimeInMinutes,
                                                                  int retentionSizeInMB) {
-        String namespace = this.testTenant + "/namespace-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
         try {
             RetentionPolicies retention = new RetentionPolicies(retentionTimeInMinutes, retentionSizeInMB);
             policies.retention_policies = retention;

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

@@ -40,6 +40,7 @@
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
 import org.apache.bookkeeper.util.StringUtils;
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
@@ -309,10 +310,10 @@ public void testTopicWithWildCardChar() throws Exception {
 
     @Test
     public void testDeleteTopicWithMissingData() throws Exception {
-        String namespace = "prop/usc-" + System.nanoTime();
+        String namespace = BrokerTestUtil.newUniqueName("prop/usc");
         admin.namespaces().createNamespace(namespace);
 
-        String topic = namespace + "/my-topic-" + System.nanoTime();
+        String topic = BrokerTestUtil.newUniqueName(namespace + "/my-topic");
 
         @Cleanup
         PulsarClient client = PulsarClient.builder()

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

@@ -28,6 +28,7 @@
 import java.util.LinkedHashSet;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Producer;
@@ -196,7 +197,7 @@ public void testPeerClusterInReplicationClusterListChange() throws Exception {
         admin1.clusters().updatePeerClusterNames("r3", null);
 
         final String serviceUrl = pulsar3.getBrokerServiceUrl();
-        final String namespace1 = "pulsar/global/peer-change-repl-ns-" + System.nanoTime();
+        final String namespace1 = BrokerTestUtil.newUniqueName("pulsar/global/peer-change-repl-ns");
         admin1.namespaces().createNamespace(namespace1);
         // add replication cluster
         admin1.namespaces().setNamespaceReplicationClusters(namespace1, Sets.newHashSet("r1"));

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

@@ -34,6 +34,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
 import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -277,7 +278,7 @@ public void testSimpleConsumerEventsWithPartition() throws Exception {
 
         int numPartitions = 4;
 
-        final String topicName = "persistent://prop/use/ns-abc/testSimpleConsumerEventsWithPartition-" + System.nanoTime();
+        final String topicName = BrokerTestUtil.newUniqueName("persistent://prop/use/ns-abc/testSimpleConsumerEventsWithPartition");
         final TopicName destName = TopicName.get(topicName);
         final String subName = "sub1";
         final int numMsgs = 100;