lhotari commented on a change in pull request #13602:
URL: https://github.com/apache/pulsar/pull/13602#discussion_r778101328
##########
File path:
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
##########
@@ -1367,6 +1369,44 @@ private void checkListContainExpectedTopic(PulsarAdmin
admin, String namespace,
}
}
+ @Test
+ private void testCloseReplicatorBeforeConnected() throws Exception {
+ String cluster = "tmp-cluster";
+ String tenant = "tmp-tenant";
+ String namespace = "vv-ns";
+ TopicName topicName = TopicName.get("persistent", tenant, namespace,
"testCloseReplicatorBeforeConnected");
+
+ // 1.Remote cluster is not available.
+ admin1.clusters().createCluster(cluster, ClusterData.builder()
+ .serviceUrl("http://localhost:18787";)
+ .brokerServiceUrl("pulsar://localhost:18787")
+ .build());
+ admin1.tenants().createTenant("tmp-tenant",
+ new TenantInfoImpl(Sets.newHashSet("appid1"),
Sets.newHashSet("r1", cluster)));
+ admin1.namespaces().createNamespace(topicName.getNamespace());
+
admin1.namespaces().setNamespaceReplicationClusters(topicName.getNamespace(),
Sets.newHashSet("r1", cluster));
+
+ // 2.Start replicator.
+ @Cleanup
+ MessageProducer producer = new MessageProducer(url1, topicName);
+ PersistentTopic topic = (PersistentTopic)
pulsar1.getBrokerService().getTopicReference(topicName.toString()).get();
+ Replicator replicator = topic.getPersistentReplicator(cluster);
+
+ // 3.Disconnect replicator.
+ Field field = AbstractReplicator.class.getDeclaredField("backOff");
+ Backoff backOff = (Backoff) field.get(replicator);
+ while (backOff.getNext() == 100) {
+ Thread.sleep(10);
+ }
+ replicator.disconnect();
+
+ // 4.Check if start task canceled.
+ long lastNum = backOff.getNext();
+ Thread.sleep(10000);
+ long curNum = backOff.getNext();
Review comment:
Consider using Awaitility instead
##########
File path:
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
##########
@@ -1367,6 +1369,44 @@ private void checkListContainExpectedTopic(PulsarAdmin
admin, String namespace,
}
}
+ @Test
+ private void testCloseReplicatorBeforeConnected() throws Exception {
+ String cluster = "tmp-cluster";
+ String tenant = "tmp-tenant";
+ String namespace = "vv-ns";
+ TopicName topicName = TopicName.get("persistent", tenant, namespace,
"testCloseReplicatorBeforeConnected");
+
+ // 1.Remote cluster is not available.
+ admin1.clusters().createCluster(cluster, ClusterData.builder()
+ .serviceUrl("http://localhost:18787";)
+ .brokerServiceUrl("pulsar://localhost:18787")
+ .build());
+ admin1.tenants().createTenant("tmp-tenant",
+ new TenantInfoImpl(Sets.newHashSet("appid1"),
Sets.newHashSet("r1", cluster)));
+ admin1.namespaces().createNamespace(topicName.getNamespace());
+
admin1.namespaces().setNamespaceReplicationClusters(topicName.getNamespace(),
Sets.newHashSet("r1", cluster));
+
+ // 2.Start replicator.
+ @Cleanup
+ MessageProducer producer = new MessageProducer(url1, topicName);
+ PersistentTopic topic = (PersistentTopic)
pulsar1.getBrokerService().getTopicReference(topicName.toString()).get();
+ Replicator replicator = topic.getPersistentReplicator(cluster);
+
+ // 3.Disconnect replicator.
+ Field field = AbstractReplicator.class.getDeclaredField("backOff");
+ Backoff backOff = (Backoff) field.get(replicator);
+ while (backOff.getNext() == 100) {
Review comment:
Please use Awaitility in tests for this type of checks
--
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]
