[Replicated] Add the test for topic replicated and create producer.
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -947,6 +948,51 @@ public class ReplicatorTest extends ReplicatorTestBase {
client1.close();
client2.close();
}
+
+ @DataProvider(name = "topicPrefix")
+ public static Object[][] topicPrefix() {
+ return new Object[][] { { "persistent://" , "/persistent" }, { "non-persistent://" , "/non-persistent" } };
+ }
+
+ @Test(dataProvider = "topicPrefix")
+ public void testTopicReplicatedAndProducerCreate(String topicPrefix, String topicName) throws Exception {
+ log.info("--- Starting ReplicatorTest::testTopicReplicatedAndProducerCreate ---");
+
+ final String cluster1 = pulsar1.getConfig().getClusterName();
+ final String cluster2 = pulsar2.getConfig().getClusterName();
+ final String namespace = "pulsar/ns-" + System.nanoTime();
+ final String partitionedTopicName = topicPrefix + namespace + topicName + "-partitioned";
+ final String nonPartitionedTopicName = topicPrefix + namespace + topicName + "-non-partitioned";
+ final int startPartitions = 4;
+ admin1.namespaces().createNamespace(namespace, Sets.newHashSet(cluster1, cluster2));
+ admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3"));
+ admin1.topics().createPartitionedTopic(partitionedTopicName, startPartitions);
+ admin1.topics().createNonPartitionedTopic(nonPartitionedTopicName);
+
+ PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString()).statsInterval(0, TimeUnit.SECONDS)
+ .build();
+ PulsarClient client2 = PulsarClient.builder().serviceUrl(url2.toString()).statsInterval(0, TimeUnit.SECONDS)
+ .build();
+
+ //persistent topic
+ Producer<byte[]> persistentProducer1 = client1.newProducer().topic(partitionedTopicName).create();
+ Producer<byte[]> persistentProducer2 = client2.newProducer().topic(partitionedTopicName).create();
+ assertNotNull(persistentProducer1.send("test".getBytes()));
+ assertNotNull(persistentProducer2.send("test".getBytes()));
+ //non-persistent topic
+ Producer<byte[]> nonPersistentProducer1 = client1.newProducer().topic(nonPartitionedTopicName).create();
+ Producer<byte[]> nonPersistentProducer2 = client2.newProducer().topic(nonPartitionedTopicName).create();
+
+ assertNotNull(nonPersistentProducer1.send("test".getBytes()));
+ assertNotNull(nonPersistentProducer2.send("test".getBytes()));
+
+ persistentProducer1.close();
+ persistentProducer2.close();
+
+ nonPersistentProducer1.close();
+ nonPersistentProducer2.close();
+ }
+
private static final Logger log = LoggerFactory.getLogger(ReplicatorTest.class);
}
