nodece commented on code in PR #24652:
URL: https://github.com/apache/pulsar/pull/24652#discussion_r2558443390
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java:
##########
@@ -1858,4 +1861,162 @@ public void
testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished() thr
// Verify: all inflight tasks are done.
ensureNoBacklogByInflightTask(getReplicator(topicName));
}
+
+ @DataProvider
+ public Object[] isPartitioned() {
+ return new Object[]{
+ true,
+ false
+ };
+ }
+
+ @Test(dataProvider = "isPartitioned")
+ public void testReplicatorCreateTopic(boolean isPartitioned) throws
Exception {
+ String ns = defaultTenant + "/" +
UUID.randomUUID().toString().replace("-", "");
+ admin1.namespaces().createNamespace(ns);
+ if (!usingGlobalZK){
+ admin2.namespaces().createNamespace(ns);
+ }
+
+ final String tp = BrokerTestUtil.newUniqueName("persistent://" + ns +
"/tp_");
+ if (isPartitioned) {
+ admin1.topics().createPartitionedTopic(tp, 4);
+ } else {
+ admin1.topics().createNonPartitionedTopic(tp);
+ }
+
+ admin1.namespaces().setNamespaceReplicationClusters(ns, new
HashSet<>(Arrays.asList(cluster1, cluster2)));
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic =
+ (PersistentTopic) broker1.getTopic(
+ isPartitioned ?
TopicName.get(tp).getPartition(0).toString() : tp, false).join()
+ .get();
+ assertFalse(persistentTopic.getReplicators().isEmpty());
+ });
+
+ @Cleanup
+ Producer<String> p1 =
client1.newProducer(Schema.STRING).topic(tp).create();
+ p1.send("msg-1");
+
+ Awaitility.await().untilAsserted(() -> {
+ if (isPartitioned){
+
assertThat(admin2.topics().getPartitionedTopicList(ns)).contains(tp);
+ }else {
+ assertThat(admin2.topics().getList(ns)).contains(tp);
+ }
+ });
+ }
+
+ @Test
+ public void
testReplicatorCreateTopicWhenTopicExistsWithDifferentTypeAcrossClusters()
throws Exception {
+ String ns = defaultTenant + "/" +
UUID.randomUUID().toString().replace("-", "");
+ admin1.namespaces().createNamespace(ns);
+ admin2.namespaces().createNamespace(ns);
+
+ final String tp = BrokerTestUtil.newUniqueName("persistent://" + ns +
"/tp_");
+ admin1.topics().createPartitionedTopic(tp, 4);
+ admin2.topics().createNonPartitionedTopic(tp);
+
+ admin1.namespaces().setNamespaceReplicationClusters(ns, new
HashSet<>(Arrays.asList(cluster1, cluster2)));
+ admin2.namespaces().setNamespaceReplicationClusters(ns, new
HashSet<>(Arrays.asList(cluster1, cluster2)));
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic =
+ (PersistentTopic)
broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join()
+ .get();
+ assertFalse(persistentTopic.getReplicators().isEmpty());
+ });
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic = (PersistentTopic)
broker2.getTopic(tp, false).join().get();
+ assertFalse(persistentTopic.getReplicators().isEmpty());
+ });
+
+ @Cleanup
+ Producer<String> p1 =
client1.newProducer(Schema.STRING).topic(tp).create();
+ p1.send("msg-p1-1");
+ @Cleanup
+ Producer<String> p2 =
client2.newProducer(Schema.STRING).topic(tp).create();
+ p2.send("msg-p2-1");
+
+ // The topic exists, but its type differs between the local and remote
clusters. The replicator should not
+ // recreate the topic.
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic =
+ (PersistentTopic)
broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join()
+ .get();
+ persistentTopic.getReplicators().forEach((key, value) -> {
+ assertFalse(value.isConnected());
+ });
+ });
+
assertThat(admin2.topics().getPartitionedTopicList(ns)).doesNotContain(tp);
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic = (PersistentTopic)
broker2.getTopic(tp, false).join().get();
+ persistentTopic.getReplicators().forEach((key, value) -> {
+ assertFalse(value.isConnected());
+ });
+ });
+ assertThat(admin1.topics().getList(ns)).doesNotContain(tp);
+ }
+
+ @Test
+ public void testReplicatorWhenPartitionCountsDiffer() throws Exception {
+ String ns = defaultTenant + "/" +
UUID.randomUUID().toString().replace("-", "");
+ admin1.namespaces().createNamespace(ns);
+ admin2.namespaces().createNamespace(ns);
+
+ final String tp = BrokerTestUtil.newUniqueName("persistent://" + ns +
"/tp_");
+ admin1.topics().createPartitionedTopic(tp, 4);
+ admin2.topics().createPartitionedTopic(tp, 8);
+
+ admin1.namespaces().setNamespaceReplicationClusters(ns, new
HashSet<>(Arrays.asList(cluster1, cluster2)));
+ admin2.namespaces().setNamespaceReplicationClusters(ns, new
HashSet<>(Arrays.asList(cluster1, cluster2)));
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic =
+ (PersistentTopic)
broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join()
+ .get();
+ assertFalse(persistentTopic.getReplicators().isEmpty());
+ });
+
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic =
+ (PersistentTopic)
broker2.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join()
+ .get();
+ assertFalse(persistentTopic.getReplicators().isEmpty());
+ });
+
+ // Trigger the replicator.
+ @Cleanup
+ Producer<String> p1 =
client1.newProducer(Schema.STRING).topic(tp).create();
+ p1.send("msg-p1-1");
+ @Cleanup
+ Producer<String> p2 =
client2.newProducer(Schema.STRING).topic(tp).create();
+ p2.send("msg-p2-1");
+
+ // Topic partition counts differ between the local and remote clusters.
+ // The replicator should not replicate the messages.
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopic persistentTopic =
+ (PersistentTopic)
broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join()
+ .get();
+ persistentTopic.getReplicators().forEach((key, value) -> {
+ assertFalse(value.isConnected());
+ });
+ });
+
+ @Cleanup
+ Consumer<String> c2 =
client2.newConsumer(Schema.STRING).topic(tp).subscriptionName("test-sub")
+
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+
+ while (true) {
+ Message<String> receive = c2.receive(3, TimeUnit.SECONDS);
+ if (receive == null) {
+ break;
+ }
+ assertEquals(receive.getValue(), "msg-p2-1");
+ }
+ }
Review Comment:
This case has been added to `testReplicatorWhenPartitionCountsDiffer`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
