[fix][flaky-test] testAutoSubscribePatternConsumer
@@ -469,17 +469,23 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
.create();
+ List<String> topicNames = Lists.newArrayList(topicName1, topicName2, topicName3);
+ NamespaceService nss = pulsar.getNamespaceService();
+ doReturn(CompletableFuture.completedFuture(topicNames)).when(nss)
+ .getListOfPersistentTopics(NamespaceName.get("my-property/my-ns"));
+
// 5. call recheckTopics to subscribe each added topics above
log.debug("recheck topics change");
PatternMultiTopicsConsumerImpl<byte[]> consumer1 = ((PatternMultiTopicsConsumerImpl<byte[]>) consumer);
consumer1.run(consumer1.getRecheckPatternTimeout());
- Thread.sleep(100);
// 6. verify consumer get methods, to get number of partitions and topics, value 6=1+2+3.
+ Awaitility.await().untilAsserted(() -> {
assertSame(pattern, ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern());
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitions().size(), 6);
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getConsumers().size(), 6);
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 2);
+ });
// 7. produce data
@@ -584,6 +590,11 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
.create();
+ List<String> topicNames = Lists.newArrayList(topicName1, topicName2, topicName3, topicName4);
+ NamespaceService nss = pulsar.getNamespaceService();
+ doReturn(CompletableFuture.completedFuture(topicNames)).when(nss)
+ .getListOfPersistentTopics(NamespaceName.get("my-property/my-ns"));
+
// 7. call recheckTopics to subscribe each added topics above, verify topics number: 10=1+2+3+4
log.debug("recheck topics change");
PatternMultiTopicsConsumerImpl<byte[]> consumer1 = ((PatternMultiTopicsConsumerImpl<byte[]>) consumer);
