You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2024/07/11 13:46:15 UTC
(pulsar) 01/02: [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted (#22854)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2b4788d6fe0c60267c3323614f9ce71beb1d4635
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue Jul 9 08:53:54 2024 +0800
[fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted (#22854)
(cherry picked from commit 9626e7e090e9481e12441a47cf7e89f209aadd03)
---
.../apache/pulsar/broker/service/TopicGCTest.java | 291 ++++++++++++++++++++-
.../client/api/PatternMultiTopicsConsumerTest.java | 37 +++
.../client/impl/PatternTopicsConsumerImplTest.java | 13 +-
.../pulsar/client/impl/TopicsConsumerImplTest.java | 5 +-
.../apache/pulsar/client/impl/ConsumerBase.java | 6 +
.../apache/pulsar/client/impl/LookupService.java | 9 +-
.../client/impl/MultiTopicsConsumerImpl.java | 188 +++++++------
.../client/impl/PatternConsumerUpdateQueue.java | 254 ++++++++++++++++++
.../impl/PatternMultiTopicsConsumerImpl.java | 249 ++++++++++++------
.../pulsar/client/impl/PulsarClientImpl.java | 15 +-
.../pulsar/client/impl/TopicListWatcher.java | 16 +-
.../impl/PatternConsumerUpdateQueueTest.java | 247 +++++++++++++++++
.../impl/PatternMultiTopicsConsumerImplTest.java | 6 +-
.../pulsar/client/impl/TopicListWatcherTest.java | 12 +-
.../pulsar/common/lookup/GetTopicsResult.java | 24 ++
.../org/apache/pulsar/common/naming/TopicName.java | 9 +
.../org/apache/pulsar/common/topics/TopicList.java | 15 +-
.../org/apache/pulsar/common/util/FutureUtil.java | 3 +
18 files changed, 1200 insertions(+), 199 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicGCTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicGCTest.java
index 8fdf0723ea8..172bd3702e1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicGCTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicGCTest.java
@@ -18,24 +18,34 @@
*/
package org.apache.pulsar.broker.service;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.TopicMessageId;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.PatternMultiTopicsConsumerImpl;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
@@ -58,14 +68,38 @@ public class TopicGCTest extends ProducerConsumerBase {
@EqualsAndHashCode.Include
protected void doInitConf() throws Exception {
super.doInitConf();
- this.conf.setBrokerDeleteInactiveTopicsEnabled(true);
- this.conf.setBrokerDeleteInactiveTopicsMode(InactiveTopicDeleteMode.delete_when_subscriptions_caught_up);
- this.conf.setBrokerDeleteInactiveTopicsFrequencySeconds(10);
+ conf.setBrokerDeleteInactiveTopicsEnabled(true);
+ conf.setBrokerDeleteInactiveTopicsMode(
+ InactiveTopicDeleteMode.delete_when_subscriptions_caught_up);
+ conf.setBrokerDeleteInactiveTopicsFrequencySeconds(10);
}
- @Test
- public void testCreateConsumerAfterOnePartDeleted() throws Exception {
+ private enum SubscribeTopicType {
+ MULTI_PARTITIONED_TOPIC,
+ REGEX_TOPIC;
+ }
+
+ @DataProvider(name = "subscribeTopicTypes")
+ public Object[][] subTopicTypes() {
+ return new Object[][]{
+ {SubscribeTopicType.MULTI_PARTITIONED_TOPIC},
+ {SubscribeTopicType.REGEX_TOPIC}
+ };
+ }
+
+ private void setSubscribeTopic(ConsumerBuilder consumerBuilder, SubscribeTopicType subscribeTopicType,
+ String topicName, String topicPattern) {
+ if (subscribeTopicType.equals(SubscribeTopicType.MULTI_PARTITIONED_TOPIC)) {
+ consumerBuilder.topic(topicName);
+ } else {
+ consumerBuilder.topicsPattern(Pattern.compile(topicPattern));
+ }
+ }
+
+ @Test(dataProvider = "subscribeTopicTypes", timeOut = 300 * 1000)
+ public void testRecreateConsumerAfterOnePartGc(SubscribeTopicType subscribeTopicType) throws Exception {
final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String topicPattern = "persistent://public/default/tp.*";
final String partition0 = topic + "-partition-0";
final String partition1 = topic + "-partition-1";
final String subscription = "s1";
@@ -77,8 +111,12 @@ public class TopicGCTest extends ProducerConsumerBase {
.enableBatching(false).create();
Producer<String> producer1 = pulsarClient.newProducer(Schema.STRING).topic(partition1)
.enableBatching(false).create();
- org.apache.pulsar.client.api.Consumer<String> consumer1 = pulsarClient.newConsumer(Schema.STRING).topic(topic)
- .subscriptionName(subscription).isAckReceiptEnabled(true).subscribe();
+ ConsumerBuilder<String> consumerBuilder1 = pulsarClient.newConsumer(Schema.STRING)
+ .subscriptionName(subscription)
+ .isAckReceiptEnabled(true)
+ .subscriptionType(SubscriptionType.Shared);
+ setSubscribeTopic(consumerBuilder1, subscribeTopicType, topic, topicPattern);
+ Consumer<String> consumer1 = consumerBuilder1.subscribe();
// Make consume all messages for one topic, do not consume any messages for another one.
producer0.send("1");
@@ -97,18 +135,247 @@ public class TopicGCTest extends ProducerConsumerBase {
});
// Verify that the consumer subscribed with partitioned topic can be created successful.
- Consumer<String> consumerAllPartition = pulsarClient.newConsumer(Schema.STRING).topic(topic)
- .subscriptionName(subscription).isAckReceiptEnabled(true).subscribe();
- Message<String> msg = consumerAllPartition.receive(2, TimeUnit.SECONDS);
+ ConsumerBuilder<String> consumerBuilder2 = pulsarClient.newConsumer(Schema.STRING)
+ .subscriptionName(subscription)
+ .isAckReceiptEnabled(true)
+ .subscriptionType(SubscriptionType.Shared);
+ setSubscribeTopic(consumerBuilder2, subscribeTopicType, topic, topicPattern);
+ Consumer<String> consumer2 = consumerBuilder2.subscribe();
+ Message<String> msg = consumer2.receive(2, TimeUnit.SECONDS);
+ String receivedMsgValue = msg.getValue();
+ log.info("received msg: {}", receivedMsgValue);
+ consumer2.acknowledge(msg);
+
+ // cleanup.
+ consumer2.close();
+ producer0.close();
+ producer1.close();
+ admin.topics().deletePartitionedTopic(topic);
+ }
+
+ @Test(dataProvider = "subscribeTopicTypes", timeOut = 300 * 1000)
+ public void testAppendCreateConsumerAfterOnePartGc(SubscribeTopicType subscribeTopicType) throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String topicPattern = "persistent://public/default/tp.*";
+ final String partition0 = topic + "-partition-0";
+ final String partition1 = topic + "-partition-1";
+ final String subscription = "s1";
+ admin.topics().createPartitionedTopic(topic, 2);
+ admin.topics().createSubscription(topic, subscription, MessageId.earliest);
+
+ // create consumers and producers.
+ Producer<String> producer0 = pulsarClient.newProducer(Schema.STRING).topic(partition0)
+ .enableBatching(false).create();
+ Producer<String> producer1 = pulsarClient.newProducer(Schema.STRING).topic(partition1)
+ .enableBatching(false).create();
+ ConsumerBuilder<String> consumerBuilder1 = pulsarClient.newConsumer(Schema.STRING)
+ .subscriptionName(subscription)
+ .isAckReceiptEnabled(true)
+ .subscriptionType(SubscriptionType.Shared);
+ setSubscribeTopic(consumerBuilder1, subscribeTopicType, topic, topicPattern);
+ Consumer<String> consumer1 = consumerBuilder1.subscribe();
+
+ // Make consume all messages for one topic, do not consume any messages for another one.
+ producer0.send("partition-0-1");
+ producer1.send("partition-1-1");
+ producer1.send("partition-1-2");
+ producer1.send("partition-1-4");
+ admin.topics().skipAllMessages(partition0, subscription);
+
+ // Wait for topic GC.
+ // Partition 0 will be deleted about 20s later, left 2min to avoid flaky.
+ producer0.close();
+ Awaitility.await().atMost(2, TimeUnit.MINUTES).untilAsserted(() -> {
+ CompletableFuture<Optional<Topic>> tp1 = pulsar.getBrokerService().getTopic(partition0, false);
+ CompletableFuture<Optional<Topic>> tp2 = pulsar.getBrokerService().getTopic(partition1, false);
+ assertTrue(tp1 == null || !tp1.get().isPresent());
+ assertTrue(tp2 != null && tp2.get().isPresent());
+ });
+
+ // Verify that the messages under "partition-1" still can be ack.
+ for (int i = 0; i < 2; i++) {
+ Message<String> msg = consumer1.receive(2, TimeUnit.SECONDS);
+ assertNotNull(msg, "Expected at least received 2 messages.");
+ log.info("received msg[{}]: {}", i, msg.getValue());
+ TopicMessageId messageId = (TopicMessageId) msg.getMessageId();
+ if (messageId.getOwnerTopic().equals(partition1)) {
+ consumer1.acknowledgeAsync(msg);
+ }
+ }
+ consumer1.close();
+
+ // Verify that the consumer subscribed with partitioned topic can be created successful.
+ ConsumerBuilder<String> consumerBuilder2 = pulsarClient.newConsumer(Schema.STRING)
+ .subscriptionName(subscription)
+ .isAckReceiptEnabled(true)
+ .subscriptionType(SubscriptionType.Shared);
+ setSubscribeTopic(consumerBuilder2, subscribeTopicType, topic, topicPattern);
+ Consumer<String> consumer2 = consumerBuilder2.subscribe();
+ producer1.send("partition-1-5");
+ Message<String> msg = consumer2.receive(2, TimeUnit.SECONDS);
assertNotNull(msg);
String receivedMsgValue = msg.getValue();
log.info("received msg: {}", receivedMsgValue);
- consumerAllPartition.acknowledge(msg);
+ consumer2.acknowledge(msg);
// cleanup.
- consumerAllPartition.close();
+ consumer2.close();
producer0.close();
producer1.close();
admin.topics().deletePartitionedTopic(topic);
}
+
+ @Test(timeOut = 180 * 1000)
+ public void testPhasePartDeletion() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String topicPattern = "persistent://public/default/tp.*";
+ final String partition0 = topic + "-partition-0";
+ final String partition1 = topic + "-partition-1";
+ final String partition2 = topic + "-partition-2";
+ final String subscription = "s1";
+ admin.topics().createPartitionedTopic(topic, 3);
+ // Create consumer.
+ PatternMultiTopicsConsumerImpl<String> c1 = (PatternMultiTopicsConsumerImpl<String>) pulsarClient
+ .newConsumer(Schema.STRING)
+ .subscriptionName(subscription)
+ .isAckReceiptEnabled(true)
+ .subscriptionType(SubscriptionType.Shared)
+ .topicsPattern(Pattern.compile(topicPattern)).subscribe();
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 1);
+ assertEquals(partitionedTopics.get(topic), 3);
+ assertEquals(consumers.size(), 3);
+ assertTrue(consumers.containsKey(partition0));
+ assertTrue(consumers.containsKey(partition1));
+ assertTrue(consumers.containsKey(partition2));
+ });
+ // Delete partitions the first time.
+ admin.topics().delete(partition0, true);
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 1);
+ assertEquals(partitionedTopics.get(topic), 3);
+ assertEquals(consumers.size(), 2);
+ assertTrue(consumers.containsKey(partition1));
+ assertTrue(consumers.containsKey(partition2));
+ });
+ // Delete partitions the second time.
+ admin.topics().delete(partition1, true);
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 1);
+ assertEquals(partitionedTopics.get(topic), 3);
+ assertEquals(consumers.size(), 1);
+ assertTrue(consumers.containsKey(partition2));
+ });
+ // Delete partitions the third time.
+ admin.topics().delete(partition2, true);
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 0);
+ assertEquals(consumers.size(), 0);
+ });
+
+ // cleanup.
+ c1.close();
+ admin.topics().deletePartitionedTopic(topic);
+ }
+
+ @Test(timeOut = 180 * 1000)
+ public void testExpandPartitions() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String topicPattern = "persistent://public/default/tp.*";
+ final String partition0 = topic + "-partition-0";
+ final String partition1 = topic + "-partition-1";
+ final String subscription = "s1";
+ admin.topics().createPartitionedTopic(topic, 2);
+ // Delete partitions.
+ admin.topics().delete(partition0, true);
+ admin.topics().delete(partition1, true);
+ // Create consumer.
+ PatternMultiTopicsConsumerImpl<String> c1 = (PatternMultiTopicsConsumerImpl<String>) pulsarClient
+ .newConsumer(Schema.STRING)
+ .subscriptionName(subscription)
+ .isAckReceiptEnabled(true)
+ .subscriptionType(SubscriptionType.Shared)
+ .topicsPattern(Pattern.compile(topicPattern)).subscribe();
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 0);
+ assertEquals(consumers.size(), 0);
+ });
+ // Trigger partitions creation.
+ pulsarClient.newConsumer(Schema.STRING).subscriptionName(subscription)
+ .subscriptionType(SubscriptionType.Shared).topic(topic).subscribe().close();
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 1);
+ assertEquals(partitionedTopics.get(topic), 2);
+ assertEquals(consumers.size(), 2);
+ assertTrue(consumers.containsKey(partition0));
+ assertTrue(consumers.containsKey(partition1));
+ });
+ // Expand partitions the first time.
+ admin.topics().updatePartitionedTopic(topic, 3);
+ final String partition2 = topic + "-partition-2";
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 1);
+ assertEquals(partitionedTopics.get(topic), 3);
+ assertEquals(consumers.size(), 3);
+ assertTrue(consumers.containsKey(partition0));
+ assertTrue(consumers.containsKey(partition1));
+ assertTrue(consumers.containsKey(partition2));
+ });
+ // Expand partitions the second time.
+ admin.topics().updatePartitionedTopic(topic, 4);
+ final String partition3 = topic + "-partition-3";
+ // Check subscriptions.
+ Awaitility.await().untilAsserted(() -> {
+ ConcurrentHashMap<String, ConsumerImpl<?>> consumers
+ = WhiteboxImpl.getInternalState(c1, "consumers");
+ ConcurrentHashMap<String, Integer> partitionedTopics
+ = WhiteboxImpl.getInternalState(c1, "partitionedTopics");
+ assertEquals(partitionedTopics.size(), 1);
+ assertEquals(partitionedTopics.get(topic), 4);
+ assertEquals(consumers.size(), 4);
+ assertTrue(consumers.containsKey(partition0));
+ assertTrue(consumers.containsKey(partition1));
+ assertTrue(consumers.containsKey(partition2));
+ assertTrue(consumers.containsKey(partition3));
+ });
+
+ // cleanup.
+ c1.close();
+ admin.topics().deletePartitionedTopic(topic);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternMultiTopicsConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternMultiTopicsConsumerTest.java
index 00a47c39571..475477ac521 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternMultiTopicsConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternMultiTopicsConsumerTest.java
@@ -18,11 +18,14 @@
*/
package org.apache.pulsar.client.api;
+import static org.testng.Assert.fail;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.impl.PatternMultiTopicsConsumerImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -95,4 +98,38 @@ public class PatternMultiTopicsConsumerTest extends ProducerConsumerBase {
consumer.close();
}
+ @Test(timeOut = 30000)
+ public void testFailedSubscribe() throws Exception {
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://public/default/tp_test");
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://public/default/tp_test");
+ final String topicName3 = BrokerTestUtil.newUniqueName("persistent://public/default/tp_test");
+ final String subName = "s1";
+ admin.topics().createPartitionedTopic(topicName1, 2);
+ admin.topics().createPartitionedTopic(topicName2, 3);
+ admin.topics().createNonPartitionedTopic(topicName3);
+
+ // Register a exclusive consumer to makes the pattern consumer failed to subscribe.
+ Consumer c1 = pulsarClient.newConsumer(Schema.STRING).topic(topicName3).subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName(subName).subscribe();
+
+ try {
+ PatternMultiTopicsConsumerImpl<String> consumer =
+ (PatternMultiTopicsConsumerImpl<String>) pulsarClient.newConsumer(Schema.STRING)
+ .topicsPattern("persistent://public/default/tp_test.*")
+ .subscriptionType(SubscriptionType.Failover)
+ .subscriptionName(subName)
+ .subscribe();
+ fail("Expected a consumer busy error.");
+ } catch (Exception ex) {
+ log.info("consumer busy", ex);
+ }
+
+ c1.close();
+ // Verify all internal consumer will be closed.
+ // If delete topic without "-f" work, it means the internal consumers were closed.
+ admin.topics().delete(topicName3);
+ admin.topics().deletePartitionedTopic(topicName2);
+ admin.topics().deletePartitionedTopic(topicName1);
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java
index 94d78e418ab..3bfecae23a5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java
@@ -19,7 +19,6 @@
package org.apache.pulsar.client.impl;
import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -35,7 +34,6 @@ import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.IntStream;
-import io.netty.util.Timeout;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.client.api.Consumer;
@@ -53,6 +51,7 @@ import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
@@ -1024,17 +1023,17 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
// 6. remove producer 1,3; verify only consumer 2 left
// seems no direct way to verify auto-unsubscribe, because this patternConsumer also referenced the topic.
- List<String> topicNames = Lists.newArrayList(topicName2);
+ String tp2p0 = TopicName.get(topicName2).getPartition(0).toString();
+ String tp2p1 = TopicName.get(topicName2).getPartition(1).toString();
+ List<String> topicNames = Lists.newArrayList(tp2p0, tp2p1);
NamespaceService nss = pulsar.getNamespaceService();
doReturn(CompletableFuture.completedFuture(topicNames)).when(nss)
.getListOfPersistentTopics(NamespaceName.get("my-property/my-ns"));
// 7. call recheckTopics to unsubscribe topic 1,3, verify topics number: 2=6-1-3
log.debug("recheck topics change");
- PatternMultiTopicsConsumerImpl<byte[]> consumer1 = ((PatternMultiTopicsConsumerImpl<byte[]>) consumer);
- Timeout recheckPatternTimeout = spy(consumer1.getRecheckPatternTimeout());
- doReturn(false).when(recheckPatternTimeout).isCancelled();
- consumer1.run(recheckPatternTimeout);
+ PatternConsumerUpdateQueue taskQueue = WhiteboxImpl.getInternalState(consumer, "updateTaskQueue");
+ taskQueue.appendRecheckOp();
Thread.sleep(100);
assertEquals(((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getPartitions().size(), 2);
assertEquals(((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getConsumers().size(), 2);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
index c343ab0d6e2..83cb5f2a440 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.client.impl;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import io.netty.util.Timeout;
+import java.time.Duration;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
@@ -1321,7 +1322,6 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
Assert.assertEquals(consumer.allTopicPartitionsNumber.intValue(), 2);
admin.topics().updatePartitionedTopic(topicName0, 5);
- consumer.getPartitionsAutoUpdateTimeout().task().run(consumer.getPartitionsAutoUpdateTimeout());
Awaitility.await().untilAsserted(() -> {
Assert.assertEquals(consumer.getPartitionsOfTheTopicMap(), 5);
@@ -1341,9 +1341,8 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
});
admin.topics().updatePartitionedTopic(topicName1, 5);
- consumer.getPartitionsAutoUpdateTimeout().task().run(consumer.getPartitionsAutoUpdateTimeout());
- Awaitility.await().untilAsserted(() -> {
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
Assert.assertEquals(consumer.getPartitionsOfTheTopicMap(), 10);
Assert.assertEquals(consumer.allTopicPartitionsNumber.intValue(), 10);
});
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 05081dcaa07..74abb82bfe8 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Queues;
import io.netty.util.Timeout;
import java.nio.charset.StandardCharsets;
@@ -1285,5 +1286,10 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
return batchReceiveTimeout != null;
}
+ @VisibleForTesting
+ CompletableFuture<Consumer<T>> getSubscribeFuture() {
+ return subscribeFuture;
+ }
+
private static final Logger log = LoggerFactory.getLogger(ConsumerBase.class);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java
index ccd1f6b23f2..2fe457059c1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java
@@ -117,7 +117,14 @@ public interface LookupService extends AutoCloseable {
InetSocketAddress resolveHost();
/**
- * Returns all the topics name for a given namespace.
+ * Returns all the topics that matches {@param topicPattern} for a given namespace.
+ *
+ * Note: {@param topicPattern} it relate to the topic name(without the partition suffix). For example:
+ * - There is a partitioned topic "tp-a" with two partitions.
+ * - tp-a-partition-0
+ * - tp-a-partition-1
+ * - If {@param topicPattern} is "tp-a", the consumer will subscribe to the two partitions.
+ * - if {@param topicPattern} is "tp-a-partition-0", the consumer will subscribe nothing.
*
* @param namespace : namespace-name
* @return
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 8047e05351a..e8cbf71e500 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -47,6 +47,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.annotation.Nullable;
@@ -68,6 +70,7 @@ import org.apache.pulsar.client.impl.transaction.TransactionImpl;
import org.apache.pulsar.client.util.ConsumerName;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.api.proto.CommandAck.AckType;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.util.CompletableFutureCancellationHandler;
@@ -81,14 +84,14 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
public static final String DUMMY_TOPIC_NAME_PREFIX = "MultiTopicsConsumer-";
// Map <topic+partition, consumer>, when get do ACK, consumer will by find by topic name
- private final ConcurrentHashMap<String, ConsumerImpl<T>> consumers;
+ protected final ConcurrentHashMap<String, ConsumerImpl<T>> consumers;
// Map <topic, numPartitions>, store partition number for each topic
protected final ConcurrentHashMap<String, Integer> partitionedTopics;
// Queue of partition consumers on which we have stopped calling receiveAsync() because the
// shared incoming queue was full
- private final ConcurrentLinkedQueue<ConsumerImpl<T>> pausedConsumers;
+ protected final ConcurrentLinkedQueue<ConsumerImpl<T>> pausedConsumers;
// sum of topicPartitions, simple topic has 1, partitioned topic equals to partition number.
AtomicInteger allTopicPartitionsNumber;
@@ -1009,8 +1012,12 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
new PulsarClientException.AlreadyClosedException("Topic name not valid"));
}
String fullTopicName = topicNameInstance.toString();
- if (consumers.containsKey(fullTopicName)
- || partitionedTopics.containsKey(topicNameInstance.getPartitionedTopicName())) {
+ if (consumers.containsKey(fullTopicName)) {
+ return FutureUtil.failedFuture(
+ new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName));
+ }
+ if (!topicNameInstance.isPartitioned()
+ && partitionedTopics.containsKey(topicNameInstance.getPartitionedTopicName())) {
return FutureUtil.failedFuture(
new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName));
}
@@ -1047,7 +1054,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
log.debug("Subscribe to topic {} metadata.partitions: {}", topicName, numPartitions);
}
- List<CompletableFuture<Consumer<T>>> futureList;
+ CompletableFuture<Void> subscribeAllPartitionsFuture;
if (numPartitions != PartitionedTopicMetadata.NON_PARTITIONED) {
// Below condition is true if subscribeAsync() has been invoked second time with same
// topicName before the first invocation had reached this point.
@@ -1067,30 +1074,50 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
ConsumerConfigurationData<T> configurationData = getInternalConsumerConfig();
configurationData.setReceiverQueueSize(receiverQueueSize);
- futureList = IntStream
- .range(0, numPartitions)
- .mapToObj(
- partitionIndex -> {
- String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
- CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
- configurationData.setStartPaused(paused);
- ConsumerImpl<T> newConsumer = createInternalConsumer(configurationData, partitionName,
- partitionIndex, subFuture, createIfDoesNotExist, schema);
- synchronized (pauseMutex) {
- if (paused) {
- newConsumer.pause();
- } else {
- newConsumer.resume();
- }
- consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
+ CompletableFuture<List<Integer>> partitionsFuture;
+ if (createIfDoesNotExist || !TopicName.get(topicName).isPersistent()) {
+ partitionsFuture = CompletableFuture.completedFuture(IntStream.range(0, numPartitions)
+ .mapToObj(i -> Integer.valueOf(i))
+ .collect(Collectors.toList()));
+ } else {
+ partitionsFuture = getExistsPartitions(topicName.toString());
+ }
+ subscribeAllPartitionsFuture = partitionsFuture.thenCompose(partitions -> {
+ if (partitions.isEmpty()) {
+ partitionedTopics.remove(topicName, numPartitions);
+ return CompletableFuture.completedFuture(null);
+ }
+ List<CompletableFuture<Consumer<T>>> subscribeList = new ArrayList<>();
+ for (int partitionIndex : partitions) {
+ String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
+ CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ configurationData.setStartPaused(paused);
+ ConsumerImpl<T> newConsumer = createInternalConsumer(configurationData, partitionName,
+ partitionIndex, subFuture, createIfDoesNotExist, schema);
+ synchronized (pauseMutex) {
+ if (paused) {
+ newConsumer.pause();
+ } else {
+ newConsumer.resume();
}
- return subFuture;
- })
- .collect(Collectors.toList());
+ Consumer originalValue = consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
+ if (originalValue != null) {
+ newConsumer.closeAsync().exceptionally(ex -> {
+ log.error("[{}] [{}] Failed to close the orphan consumer",
+ partitionName, subscription, ex);
+ return null;
+ });
+ }
+ }
+ subscribeList.add(subFuture);
+ }
+ return FutureUtil.waitForAll(subscribeList);
+ });
} else {
allTopicPartitionsNumber.incrementAndGet();
- CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ CompletableFuture<Consumer<T>> subscribeFuture = new CompletableFuture<>();
+ subscribeAllPartitionsFuture = subscribeFuture.thenAccept(__ -> {});
synchronized (pauseMutex) {
consumers.compute(topicName, (key, existingValue) -> {
@@ -1104,7 +1131,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
} else {
internalConfig.setStartPaused(paused);
ConsumerImpl<T> newConsumer = createInternalConsumer(internalConfig, topicName,
- -1, subFuture, createIfDoesNotExist, schema);
+ -1, subscribeFuture, createIfDoesNotExist, schema);
if (paused) {
newConsumer.pause();
} else {
@@ -1114,11 +1141,10 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
}
});
}
- futureList = Collections.singletonList(subFuture);
+
}
- FutureUtil.waitForAll(futureList)
- .thenAccept(finalFuture -> {
+ subscribeAllPartitionsFuture.thenAccept(finalFuture -> {
if (allTopicPartitionsNumber.get() > getCurrentReceiverQueueSize()) {
setCurrentReceiverQueueSize(allTopicPartitionsNumber.get());
}
@@ -1139,6 +1165,8 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
return;
})
.exceptionally(ex -> {
+ log.warn("[{}] Failed to subscribe for topic [{}] in topics consumer {}", topic, topicName,
+ ex.getMessage());
handleSubscribeOneTopicError(topicName, ex, subscribeResult);
return null;
});
@@ -1162,7 +1190,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
}
// handling failure during subscribe new topic, unsubscribe success created partitions
- private void handleSubscribeOneTopicError(String topicName,
+ protected void handleSubscribeOneTopicError(String topicName,
Throwable error,
CompletableFuture<Void> subscribeFuture) {
log.warn("[{}] Failed to subscribe for topic [{}] in topics consumer {}", topic, topicName, error.getMessage());
@@ -1255,59 +1283,6 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
return unsubscribeFuture;
}
- /***
- * Remove a consumer for a topic.
- * @param topicName topic name contains the partition suffix.
- */
- public CompletableFuture<Void> removeConsumerAsync(String topicName) {
- checkArgument(TopicName.isValid(topicName), "Invalid topic name:" + topicName);
-
- if (getState() == State.Closing || getState() == State.Closed) {
- return FutureUtil.failedFuture(
- new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed"));
- }
-
- CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>();
- String topicPartName = TopicName.get(topicName).getPartitionedTopicName();
-
-
- List<ConsumerImpl<T>> consumersToClose = consumers.values().stream()
- .filter(consumer -> {
- String consumerTopicName = consumer.getTopic();
- return TopicName.get(consumerTopicName).getPartitionedTopicName().equals(topicPartName);
- }).collect(Collectors.toList());
-
- List<CompletableFuture<Void>> futureList = consumersToClose.stream()
- .map(ConsumerImpl::closeAsync).collect(Collectors.toList());
-
- FutureUtil.waitForAll(futureList)
- .whenComplete((r, ex) -> {
- if (ex == null) {
- consumersToClose.forEach(consumer1 -> {
- consumers.remove(consumer1.getTopic());
- pausedConsumers.remove(consumer1);
- allTopicPartitionsNumber.decrementAndGet();
- });
-
- removeTopic(topicName);
- if (unAckedMessageTracker instanceof UnAckedTopicMessageTracker) {
- ((UnAckedTopicMessageTracker) unAckedMessageTracker).removeTopicMessages(topicName);
- }
-
- unsubscribeFuture.complete(null);
- log.info("[{}] [{}] [{}] Removed Topics Consumer, allTopicPartitionsNumber: {}",
- topicName, subscription, consumerName, allTopicPartitionsNumber);
- } else {
- unsubscribeFuture.completeExceptionally(ex);
- setState(State.Failed);
- log.error("[{}] [{}] [{}] Could not remove Topics Consumer",
- topicName, subscription, consumerName, ex.getCause());
- }
- });
-
- return unsubscribeFuture;
- }
-
// get topics name
public List<String> getPartitionedTopics() {
@@ -1573,4 +1548,51 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
CURRENT_RECEIVER_QUEUE_SIZE_UPDATER.set(this, newSize);
resumeReceivingFromPausedConsumersIfNeeded();
}
+
+ /**
+ * Get the exists partitions of a partitioned topic, the result does not contain the partitions which has not been
+ * created yet(in other words, the partitions that do not exist in the response of "pulsar-admin topics list").
+ * @return sorted partitions list if it is a partitioned topic; @return an empty list if it is a non-partitioned
+ * topic.
+ */
+ private CompletableFuture<List<Integer>> getExistsPartitions(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ if (!topicName.isPersistent()) {
+ return FutureUtil.failedFuture(new IllegalArgumentException("The method getExistsPartitions"
+ + " does not support non-persistent topic yet."));
+ }
+ return client.getLookup().getTopicsUnderNamespace(topicName.getNamespaceObject(),
+ CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ TopicName.getPattern(topicName.getPartitionedTopicName()),
+ null).thenApply(getTopicsResult -> {
+ if (getTopicsResult.getNonPartitionedOrPartitionTopics() == null
+ || getTopicsResult.getNonPartitionedOrPartitionTopics().isEmpty()) {
+ return Collections.emptyList();
+ }
+ // If broker version is less than "2.11.x", it does not support broker-side pattern check, so append
+ // a client-side pattern check.
+ // If lookup service is typed HttpLookupService, the HTTP API does not support broker-side pattern
+ // check yet, so append a client-side pattern check.
+ Predicate<String> clientSideFilter;
+ if (getTopicsResult.isFiltered()) {
+ clientSideFilter = __ -> true;
+ } else {
+ clientSideFilter =
+ tp -> Pattern.compile(TopicName.getPartitionPattern(topic)).matcher(tp).matches();
+ }
+ ArrayList<Integer> list = new ArrayList<>(getTopicsResult.getNonPartitionedOrPartitionTopics().size());
+ for (String partition : getTopicsResult.getNonPartitionedOrPartitionTopics()) {
+ int partitionIndex = TopicName.get(partition).getPartitionIndex();
+ if (partitionIndex < 0) {
+ // It is not a partition.
+ continue;
+ }
+ if (clientSideFilter.test(partition)) {
+ list.add(partitionIndex);
+ }
+ }
+ Collections.sort(list);
+ return list;
+ });
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueue.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueue.java
new file mode 100644
index 00000000000..d6eba6463a0
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueue.java
@@ -0,0 +1,254 @@
+/*
+ * 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.client.impl;
+
+import com.google.common.annotations.VisibleForTesting;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * Used to make all tasks that will modify subscriptions will be executed one by one, and skip the unnecessary updating.
+ *
+ * So far, four three scenarios that will modify subscriptions:
+ * 1. When start pattern consumer.
+ * 2. After topic list watcher reconnected, it will call {@link PatternMultiTopicsConsumerImpl#recheckTopicsChange()}.
+ * this scenario only exists in the version >= 2.11 (both client-version and broker version are >= 2.11).
+ * 3. A scheduled task will call {@link PatternMultiTopicsConsumerImpl#recheckTopicsChange()}, this scenario only
+ * exists in the version < 2.11.
+ * 4. The topics change events will trigger a
+ * {@link PatternMultiTopicsConsumerImpl#topicsChangeListener#onTopicsRemoved(Collection)} or
+ * {@link PatternMultiTopicsConsumerImpl#topicsChangeListener#onTopicsAdded(Collection)}.
+ *
+ * When you are using this client connect to the broker whose version >= 2.11, there are three scenarios: [1, 2, 4].
+ * When you are using this client connect to the broker whose version < 2.11, there is only one scenario: [3] and all
+ * the event will run in the same thread.
+ */
+@Slf4j
+@SuppressFBWarnings("EI_EXPOSE_REP2")
+public class PatternConsumerUpdateQueue {
+
+ private static final Pair<UpdateSubscriptionType, Collection<String>> RECHECK_OP =
+ Pair.of(UpdateSubscriptionType.RECHECK, null);
+
+ private final LinkedBlockingQueue<Pair<UpdateSubscriptionType, Collection<String>>> pendingTasks;
+
+ private final PatternMultiTopicsConsumerImpl patternConsumer;
+
+ private final PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener;
+
+ /**
+ * Whether there is a task is in progress, this variable is used to confirm whether a next-task triggering is
+ * needed.
+ */
+ private Pair<UpdateSubscriptionType, CompletableFuture<Void>> taskInProgress = null;
+
+ /**
+ * Whether there is a recheck task in queue.
+ * - Since recheck task will do all changes, it can be used to compress multiple tasks to one.
+ * - To avoid skipping the newest changes, once the recheck task is starting to work, this variable will be set
+ * to "false".
+ */
+ private boolean recheckTaskInQueue = false;
+
+ private volatile long lastRecheckTaskStartingTimestamp = 0;
+
+ private boolean closed;
+
+ public PatternConsumerUpdateQueue(PatternMultiTopicsConsumerImpl patternConsumer) {
+ this(patternConsumer, patternConsumer.topicsChangeListener);
+ }
+
+ /** This constructor is only for test. **/
+ @VisibleForTesting
+ public PatternConsumerUpdateQueue(PatternMultiTopicsConsumerImpl patternConsumer,
+ PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener) {
+ this.patternConsumer = patternConsumer;
+ this.topicsChangeListener = topicsChangeListener;
+ this.pendingTasks = new LinkedBlockingQueue<>();
+ // To avoid subscribing and topics changed events execute concurrently, let the change events starts after the
+ // subscribing task.
+ doAppend(Pair.of(UpdateSubscriptionType.CONSUMER_INIT, null));
+ }
+
+ synchronized void appendTopicsAddedOp(Collection<String> topics) {
+ if (topics == null || topics.isEmpty()) {
+ return;
+ }
+ doAppend(Pair.of(UpdateSubscriptionType.TOPICS_ADDED, topics));
+ }
+
+ synchronized void appendTopicsRemovedOp(Collection<String> topics) {
+ if (topics == null || topics.isEmpty()) {
+ return;
+ }
+ doAppend(Pair.of(UpdateSubscriptionType.TOPICS_REMOVED, topics));
+ }
+
+ synchronized void appendRecheckOp() {
+ doAppend(RECHECK_OP);
+ }
+
+ synchronized void doAppend(Pair<UpdateSubscriptionType, Collection<String>> task) {
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] try to append task. {} {}", patternConsumer.getSubscription(),
+ task.getLeft(), task.getRight() == null ? "" : task.getRight());
+ }
+ // Once there is a recheck task in queue, it means other tasks can be skipped.
+ if (recheckTaskInQueue) {
+ return;
+ }
+
+ // Once there are too many tasks in queue, compress them as a recheck task.
+ if (pendingTasks.size() >= 30 && !task.getLeft().equals(UpdateSubscriptionType.RECHECK)) {
+ appendRecheckOp();
+ return;
+ }
+
+ pendingTasks.add(task);
+ if (task.getLeft().equals(UpdateSubscriptionType.RECHECK)) {
+ recheckTaskInQueue = true;
+ }
+
+ // If no task is in-progress, trigger a task execution.
+ if (taskInProgress == null) {
+ triggerNextTask();
+ }
+ }
+
+ synchronized void triggerNextTask() {
+ if (closed) {
+ return;
+ }
+
+ final Pair<UpdateSubscriptionType, Collection<String>> task = pendingTasks.poll();
+
+ // No pending task.
+ if (task == null) {
+ taskInProgress = null;
+ return;
+ }
+
+ // If there is a recheck task in queue, skip others and only call the recheck task.
+ if (recheckTaskInQueue && !task.getLeft().equals(UpdateSubscriptionType.RECHECK)) {
+ triggerNextTask();
+ return;
+ }
+
+ // Execute pending task.
+ CompletableFuture<Void> newTaskFuture = null;
+ switch (task.getLeft()) {
+ case CONSUMER_INIT: {
+ newTaskFuture = patternConsumer.getSubscribeFuture().thenAccept(__ -> {}).exceptionally(ex -> {
+ // If the subscribe future was failed, the consumer will be closed.
+ synchronized (PatternConsumerUpdateQueue.this) {
+ this.closed = true;
+ patternConsumer.closeAsync().exceptionally(ex2 -> {
+ log.error("Pattern consumer failed to close, this error may left orphan consumers."
+ + " Subscription: {}", patternConsumer.getSubscription());
+ return null;
+ });
+ }
+ return null;
+ });
+ break;
+ }
+ case TOPICS_ADDED: {
+ newTaskFuture = topicsChangeListener.onTopicsAdded(task.getRight());
+ break;
+ }
+ case TOPICS_REMOVED: {
+ newTaskFuture = topicsChangeListener.onTopicsRemoved(task.getRight());
+ break;
+ }
+ case RECHECK: {
+ recheckTaskInQueue = false;
+ lastRecheckTaskStartingTimestamp = System.currentTimeMillis();
+ newTaskFuture = patternConsumer.recheckTopicsChange();
+ break;
+ }
+ default: {
+ throw new RuntimeException("Un-support UpdateSubscriptionType");
+ }
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] starting task. {} {} ", patternConsumer.getSubscription(),
+ task.getLeft(), task.getRight() == null ? "" : task.getRight());
+ }
+ // Trigger next pending task.
+ taskInProgress = Pair.of(task.getLeft(), newTaskFuture);
+ newTaskFuture.thenAccept(ignore -> {
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] task finished. {} {} ", patternConsumer.getSubscription(),
+ task.getLeft(), task.getRight() == null ? "" : task.getRight());
+ }
+ triggerNextTask();
+ }).exceptionally(ex -> {
+ /**
+ * Once a updating fails, trigger a delayed new recheck task to guarantee all things is correct.
+ * - Skip if there is already a recheck task in queue.
+ * - Skip if the last recheck task has been executed after the current time.
+ */
+ log.error("Pattern consumer [{}] task finished. {} {}. But it failed", patternConsumer.getSubscription(),
+ task.getLeft(), task.getRight() == null ? "" : task.getRight(), ex);
+ // Skip if there is already a recheck task in queue.
+ synchronized (PatternConsumerUpdateQueue.this) {
+ if (recheckTaskInQueue || PatternConsumerUpdateQueue.this.closed) {
+ return null;
+ }
+ }
+ // Skip if the last recheck task has been executed after the current time.
+ long failedTime = System.currentTimeMillis();
+ patternConsumer.getClient().timer().newTimeout(timeout -> {
+ if (lastRecheckTaskStartingTimestamp <= failedTime) {
+ appendRecheckOp();
+ }
+ }, 10, TimeUnit.SECONDS);
+ triggerNextTask();
+ return null;
+ });
+ }
+
+ public synchronized CompletableFuture<Void> cancelAllAndWaitForTheRunningTask() {
+ this.closed = true;
+ if (taskInProgress == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ // If the in-progress task is consumer init task, it means nothing is in-progress.
+ if (taskInProgress.getLeft().equals(UpdateSubscriptionType.CONSUMER_INIT)) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return taskInProgress.getRight().thenAccept(__ -> {}).exceptionally(ex -> null);
+ }
+
+ private enum UpdateSubscriptionType {
+ /** A marker that indicates the consumer's subscribe task.**/
+ CONSUMER_INIT,
+ /** Triggered by {@link PatternMultiTopicsConsumerImpl#topicsChangeListener}.**/
+ TOPICS_ADDED,
+ /** Triggered by {@link PatternMultiTopicsConsumerImpl#topicsChangeListener}.**/
+ TOPICS_REMOVED,
+ /** A fully check for pattern consumer. **/
+ RECHECK;
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
index ec7ff7930c0..91c6da26d59 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
@@ -27,13 +27,13 @@ import io.netty.util.TimerTask;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
-import java.util.stream.Collectors;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
@@ -42,6 +42,7 @@ import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
import org.apache.pulsar.common.lookup.GetTopicsResult;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.topics.TopicList;
import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.BackoffBuilder;
@@ -51,7 +52,7 @@ import org.slf4j.LoggerFactory;
public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T> implements TimerTask {
private final Pattern topicsPattern;
- private final TopicsChangedListener topicsChangeListener;
+ final TopicsChangedListener topicsChangeListener;
private final Mode subscriptionMode;
private final CompletableFuture<TopicListWatcher> watcherFuture = new CompletableFuture<>();
protected NamespaceName namespaceName;
@@ -69,6 +70,8 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
private volatile Timeout recheckPatternTimeout = null;
private volatile String topicsHash;
+ private PatternConsumerUpdateQueue updateTaskQueue;
+
/***
* @param topicsPattern The regexp for the topic name(not contains partition suffix).
*/
@@ -98,20 +101,23 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
checkArgument(getNameSpaceFromPattern(topicsPattern).toString().equals(this.namespaceName.toString()));
this.topicsChangeListener = new PatternTopicsChangedListener();
+ this.updateTaskQueue = new PatternConsumerUpdateQueue(this);
this.recheckPatternTimeout = client.timer()
.newTimeout(this, Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS);
if (subscriptionMode == Mode.PERSISTENT) {
long watcherId = client.newTopicListWatcherId();
- new TopicListWatcher(topicsChangeListener, client, topicsPattern, watcherId,
+ new TopicListWatcher(updateTaskQueue, client, topicsPattern, watcherId,
namespaceName, topicsHash, watcherFuture, () -> recheckTopicsChangeAfterReconnect());
watcherFuture
.thenAccept(__ -> recheckPatternTimeout.cancel())
.exceptionally(ex -> {
- log.warn("Unable to create topic list watcher. Falling back to only polling for new topics", ex);
+ log.warn("Pattern consumer [{}] unable to create topic list watcher. Falling back to only polling"
+ + " for new topics", conf.getSubscriptionName(), ex);
return null;
});
} else {
- log.debug("Not creating topic list watcher for subscription mode {}", subscriptionMode);
+ log.debug("Pattern consumer [{}] not creating topic list watcher for subscription mode {}",
+ conf.getSubscriptionName(), subscriptionMode);
watcherFuture.complete(null);
}
}
@@ -129,17 +135,7 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
return;
}
// Do check.
- recheckTopicsChange().whenComplete((ignore, ex) -> {
- if (ex != null) {
- log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage());
- long delayMs = recheckPatternTaskBackoff.next();
- client.timer().newTimeout(timeout -> {
- recheckTopicsChangeAfterReconnect();
- }, delayMs, TimeUnit.MILLISECONDS);
- } else {
- recheckPatternTaskBackoff.reset();
- }
- });
+ updateTaskQueue.appendRecheckOp();
}
// TimerTask to recheck topics change, and trigger subscribe/unsubscribe based on the change.
@@ -148,18 +144,10 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
if (timeout.isCancelled()) {
return;
}
- recheckTopicsChange().exceptionally(ex -> {
- log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage());
- return null;
- }).thenAccept(__ -> {
- // schedule the next re-check task
- this.recheckPatternTimeout = client.timer()
- .newTimeout(PatternMultiTopicsConsumerImpl.this,
- Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS);
- });
+ updateTaskQueue.appendRecheckOp();
}
- private CompletableFuture<Void> recheckTopicsChange() {
+ CompletableFuture<Void> recheckTopicsChange() {
String pattern = topicsPattern.pattern();
final int epoch = recheckPatternEpoch.incrementAndGet();
return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash)
@@ -172,22 +160,18 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
return CompletableFuture.completedFuture(null);
}
if (log.isDebugEnabled()) {
- log.debug("Get topics under namespace {}, topics.size: {}, topicsHash: {}, filtered: {}",
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topics.size: {},"
+ + " topicsHash: {}, filtered: {}",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(),
namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(),
getTopicsResult.isFiltered());
getTopicsResult.getTopics().forEach(topicName ->
log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName));
}
- final List<String> oldTopics = new ArrayList<>(getPartitionedTopics());
- for (String partition : getPartitions()) {
- TopicName topicName = TopicName.get(partition);
- if (!topicName.isPartitioned() || !oldTopics.contains(topicName.getPartitionedTopicName())) {
- oldTopics.add(partition);
- }
- }
+ final List<String> oldTopics = new ArrayList<>(getPartitions());
return updateSubscriptions(topicsPattern, this::setTopicsHash, getTopicsResult,
- topicsChangeListener, oldTopics);
+ topicsChangeListener, oldTopics, subscription);
}
});
}
@@ -196,7 +180,8 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
java.util.function.Consumer<String> topicsHashSetter,
GetTopicsResult getTopicsResult,
TopicsChangedListener topicsChangedListener,
- List<String> oldTopics) {
+ List<String> oldTopics,
+ String subscriptionForLog) {
topicsHashSetter.accept(getTopicsResult.getTopicsHash());
if (!getTopicsResult.isChanged()) {
return CompletableFuture.completedFuture(null);
@@ -204,14 +189,20 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
List<String> newTopics;
if (getTopicsResult.isFiltered()) {
- newTopics = getTopicsResult.getTopics();
+ newTopics = getTopicsResult.getNonPartitionedOrPartitionTopics();
} else {
- newTopics = TopicList.filterTopics(getTopicsResult.getTopics(), topicsPattern);
+ newTopics = getTopicsResult.filterTopics(topicsPattern).getNonPartitionedOrPartitionTopics();
}
final List<CompletableFuture<?>> listenersCallback = new ArrayList<>(2);
- listenersCallback.add(topicsChangedListener.onTopicsAdded(TopicList.minus(newTopics, oldTopics)));
- listenersCallback.add(topicsChangedListener.onTopicsRemoved(TopicList.minus(oldTopics, newTopics)));
+ Set<String> topicsAdded = TopicList.minus(newTopics, oldTopics);
+ Set<String> topicsRemoved = TopicList.minus(oldTopics, newTopics);
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] Recheck pattern consumer's topics. topicsAdded: {}, topicsRemoved: {}",
+ subscriptionForLog, topicsAdded, topicsRemoved);
+ }
+ listenersCallback.add(topicsChangedListener.onTopicsAdded(topicsAdded));
+ listenersCallback.add(topicsChangedListener.onTopicsRemoved(topicsRemoved));
return FutureUtil.waitForAll(Collections.unmodifiableList(listenersCallback));
}
@@ -247,23 +238,68 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
*/
@Override
public CompletableFuture<Void> onTopicsRemoved(Collection<String> removedTopics) {
- CompletableFuture<Void> removeFuture = new CompletableFuture<>();
-
if (removedTopics.isEmpty()) {
- removeFuture.complete(null);
- return removeFuture;
+ return CompletableFuture.completedFuture(null);
}
- List<CompletableFuture<Void>> futures = Lists.newArrayListWithExpectedSize(partitionedTopics.size());
- removedTopics.stream().forEach(topic -> futures.add(removeConsumerAsync(topic)));
- FutureUtil.waitForAll(futures)
- .thenAccept(finalFuture -> removeFuture.complete(null))
- .exceptionally(ex -> {
- log.warn("[{}] Failed to unsubscribe from topics: {}", topic, ex.getMessage());
- removeFuture.completeExceptionally(ex);
+ // Unsubscribe and remove consumers in memory.
+ List<CompletableFuture<Void>> unsubscribeList = new ArrayList<>(removedTopics.size());
+ Set<String> partialRemoved = new HashSet<>(removedTopics.size());
+ Set<String> partialRemovedForLog = new HashSet<>(removedTopics.size());
+ for (String tp : removedTopics) {
+ TopicName topicName = TopicName.get(tp);
+ ConsumerImpl<T> consumer = consumers.get(topicName.toString());
+ if (consumer != null) {
+ CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>();
+ consumer.closeAsync().whenComplete((__, ex) -> {
+ if (ex != null) {
+ log.error("Pattern consumer [{}] failed to unsubscribe from topics: {}",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(), topicName.toString(), ex);
+ unsubscribeFuture.completeExceptionally(ex);
+ } else {
+ consumers.remove(topicName.toString(), consumer);
+ unsubscribeFuture.complete(null);
+ }
+ });
+ unsubscribeList.add(unsubscribeFuture);
+ partialRemoved.add(topicName.getPartitionedTopicName());
+ partialRemovedForLog.add(topicName.toString());
+ }
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] remove topics. {}",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(),
+ partialRemovedForLog);
+ }
+
+ // Remove partitioned topics in memory.
+ return FutureUtil.waitForAll(unsubscribeList).handle((__, ex) -> {
+ List<String> removedPartitionedTopicsForLog = new ArrayList<>();
+ for (String groupedTopicRemoved : partialRemoved) {
+ Integer partitions = partitionedTopics.get(groupedTopicRemoved);
+ if (partitions != null) {
+ boolean allPartitionsHasBeenRemoved = true;
+ for (int i = 0; i < partitions; i++) {
+ if (consumers.containsKey(
+ TopicName.get(groupedTopicRemoved).getPartition(i).toString())) {
+ allPartitionsHasBeenRemoved = false;
+ break;
+ }
+ }
+ if (allPartitionsHasBeenRemoved) {
+ removedPartitionedTopicsForLog.add(String.format("%s with %s partitions",
+ groupedTopicRemoved, partitions));
+ partitionedTopics.remove(groupedTopicRemoved, partitions);
+ }
+ }
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] remove partitioned topics because all partitions have been"
+ + " removed. {}", PatternMultiTopicsConsumerImpl.this.getSubscription(),
+ removedPartitionedTopicsForLog);
+ }
return null;
});
- return removeFuture;
}
/**
@@ -271,29 +307,90 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
*/
@Override
public CompletableFuture<Void> onTopicsAdded(Collection<String> addedTopics) {
- CompletableFuture<Void> addFuture = new CompletableFuture<>();
-
if (addedTopics.isEmpty()) {
- addFuture.complete(null);
- return addFuture;
+ return CompletableFuture.completedFuture(null);
}
-
- Set<String> addTopicPartitionedName = addedTopics.stream()
- .map(addTopicName -> TopicName.get(addTopicName).getPartitionedTopicName())
- .collect(Collectors.toSet());
-
- List<CompletableFuture<Void>> futures = Lists.newArrayListWithExpectedSize(partitionedTopics.size());
- addTopicPartitionedName.forEach(partitionedTopic -> futures.add(
- subscribeAsync(partitionedTopic,
- false /* createTopicIfDoesNotExist */)));
- FutureUtil.waitForAll(futures)
- .thenAccept(finalFuture -> addFuture.complete(null))
- .exceptionally(ex -> {
- log.warn("[{}] Failed to subscribe to topics: {}", topic, ex.getMessage());
- addFuture.completeExceptionally(ex);
- return null;
+ List<CompletableFuture<Void>> futures = Lists.newArrayListWithExpectedSize(addedTopics.size());
+ /**
+ * Three normal cases:
+ * 1. Expand partitions.
+ * 2. Non-partitioned topic, but has been subscribing.
+ * 3. Non-partitioned topic or Partitioned topic, but has not been subscribing.
+ * Two unexpected cases:
+ * Error-1: Received adding non-partitioned topic event, but has subscribed a partitioned topic with the
+ * same name.
+ * Error-2: Received adding partitioned topic event, but has subscribed a non-partitioned topic with the
+ * same name.
+ *
+ * Note: The events that triggered by {@link TopicsPartitionChangedListener} after expanding partitions has
+ * been disabled through "conf.setAutoUpdatePartitions(false)" when creating
+ * {@link PatternMultiTopicsConsumerImpl}.
+ */
+ Set<String> groupedTopics = new HashSet<>();
+ List<String> expendPartitionsForLog = new ArrayList<>();
+ for (String tp : addedTopics) {
+ TopicName topicName = TopicName.get(tp);
+ groupedTopics.add(topicName.getPartitionedTopicName());
+ }
+ for (String tp : addedTopics) {
+ TopicName topicName = TopicName.get(tp);
+ // Case 1: Expand partitions.
+ if (partitionedTopics.containsKey(topicName.getPartitionedTopicName())) {
+ if (consumers.containsKey(topicName.toString())) {
+ // Already subscribed.
+ } else if (topicName.getPartitionIndex() < 0) {
+ // Error-1: Received adding non-partitioned topic event, but has subscribed a partitioned topic
+ // with the same name.
+ log.error("Pattern consumer [{}] skip to subscribe to the non-partitioned topic {}, because has"
+ + "subscribed a partitioned topic with the same name",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(), topicName.toString());
+ } else {
+ if (topicName.getPartitionIndex() + 1
+ > partitionedTopics.get(topicName.getPartitionedTopicName())) {
+ partitionedTopics.put(topicName.getPartitionedTopicName(),
+ topicName.getPartitionIndex() + 1);
+ }
+ expendPartitionsForLog.add(topicName.toString());
+ CompletableFuture consumerFuture = subscribeAsync(topicName.toString(),
+ PartitionedTopicMetadata.NON_PARTITIONED);
+ consumerFuture.whenComplete((__, ex) -> {
+ if (ex != null) {
+ log.warn("Pattern consumer [{}] Failed to subscribe to topics: {}",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(), topicName, ex);
+ }
+ });
+ futures.add(consumerFuture);
+ }
+ groupedTopics.remove(topicName.getPartitionedTopicName());
+ } else if (consumers.containsKey(topicName.toString())) {
+ // Case-2: Non-partitioned topic, but has been subscribing.
+ groupedTopics.remove(topicName.getPartitionedTopicName());
+ } else if (consumers.containsKey(topicName.getPartitionedTopicName())
+ && topicName.getPartitionIndex() >= 0) {
+ // Error-2: Received adding partitioned topic event, but has subscribed a non-partitioned topic
+ // with the same name.
+ log.error("Pattern consumer [{}] skip to subscribe to the partitioned topic {}, because has"
+ + "subscribed a non-partitioned topic with the same name",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(), topicName);
+ groupedTopics.remove(topicName.getPartitionedTopicName());
+ }
+ }
+ // Case 3: Non-partitioned topic or Partitioned topic, which has not been subscribed.
+ for (String partitionedTopic : groupedTopics) {
+ CompletableFuture consumerFuture = subscribeAsync(partitionedTopic, false);
+ consumerFuture.whenComplete((__, ex) -> {
+ if (ex != null) {
+ log.warn("Pattern consumer [{}] Failed to subscribe to topics: {}",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(), partitionedTopic, ex);
+ }
});
- return addFuture;
+ futures.add(consumerFuture);
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] add topics. expend partitions {}, new subscribing {}",
+ PatternMultiTopicsConsumerImpl.this.getSubscription(), expendPartitionsForLog, groupedTopics);
+ }
+ return FutureUtil.waitForAll(futures);
}
}
@@ -313,7 +410,7 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
closeFutures.add(watcher.closeAsync());
}
}
- closeFutures.add(super.closeAsync());
+ closeFutures.add(updateTaskQueue.cancelAllAndWaitForTheRunningTask().thenCompose(__ -> super.closeAsync()));
return FutureUtil.waitForAll(closeFutures);
}
@@ -322,5 +419,11 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
return recheckPatternTimeout;
}
+ protected void handleSubscribeOneTopicError(String topicName,
+ Throwable error,
+ CompletableFuture<Void> subscribeFuture) {
+ subscribeFuture.completeExceptionally(error);
+ }
+
private static final Logger log = LoggerFactory.getLogger(PatternMultiTopicsConsumerImpl.class);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index b93786bb4b1..64b6e45cb7f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -583,12 +583,13 @@ public class PulsarClientImpl implements PulsarClient {
lookup.getTopicsUnderNamespace(namespaceName, subscriptionMode, regex, null)
.thenAccept(getTopicsResult -> {
if (log.isDebugEnabled()) {
- log.debug("Get topics under namespace {}, topics.size: {},"
- + " topicsHash: {}, changed: {}, filtered: {}",
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topics.size: {},"
+ + " topicsHash: {}, changed: {}, filtered: {}", conf.getSubscriptionName(),
namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(),
getTopicsResult.isChanged(), getTopicsResult.isFiltered());
getTopicsResult.getTopics().forEach(topicName ->
- log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName));
+ log.debug("Pattern consumer [{}] get topics under namespace {}, topic: {}",
+ conf.getSubscriptionName(), namespaceName, topicName));
}
List<String> topicsList = getTopicsResult.getTopics();
@@ -596,6 +597,14 @@ public class PulsarClientImpl implements PulsarClient {
topicsList = TopicList.filterTopics(getTopicsResult.getTopics(), conf.getTopicsPattern());
}
conf.getTopicNames().addAll(topicsList);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] initialize topics. {}", conf.getSubscriptionName(),
+ getTopicsResult.getNonPartitionedOrPartitionTopics());
+ }
+
+ // Pattern consumer has his unique check mechanism, so do not need the feature "autoUpdatePartitions".
+ conf.setAutoUpdatePartitions(false);
ConsumerBase<T> consumer = new PatternMultiTopicsConsumerImpl<>(conf.getTopicsPattern(),
getTopicsResult.getTopicsHash(),
PulsarClientImpl.this,
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java
index 4e635e0d2e8..9cb737bd73d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java
@@ -43,7 +43,7 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
AtomicLongFieldUpdater
.newUpdater(TopicListWatcher.class, "createWatcherDeadline");
- private final PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener;
+ private final PatternConsumerUpdateQueue patternConsumerUpdateQueue;
private final String name;
private final ConnectionHandler connectionHandler;
private final Pattern topicsPattern;
@@ -63,13 +63,13 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
/***
* @param topicsPattern The regexp for the topic name(not contains partition suffix).
*/
- public TopicListWatcher(PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener,
+ public TopicListWatcher(PatternConsumerUpdateQueue patternConsumerUpdateQueue,
PulsarClientImpl client, Pattern topicsPattern, long watcherId,
NamespaceName namespace, String topicsHash,
CompletableFuture<TopicListWatcher> watcherFuture,
Runnable recheckTopicsChangeAfterReconnect) {
super(client, topicsPattern.pattern());
- this.topicsChangeListener = topicsChangeListener;
+ this.patternConsumerUpdateQueue = patternConsumerUpdateQueue;
this.name = "Watcher(" + topicsPattern + ")";
this.connectionHandler = new ConnectionHandler(this,
new BackoffBuilder()
@@ -277,13 +277,7 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
}
public void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate update) {
- List<String> deleted = update.getDeletedTopicsList();
- if (!deleted.isEmpty()) {
- topicsChangeListener.onTopicsRemoved(deleted);
- }
- List<String> added = update.getNewTopicsList();
- if (!added.isEmpty()) {
- topicsChangeListener.onTopicsAdded(added);
- }
+ patternConsumerUpdateQueue.appendTopicsRemovedOp(update.getDeletedTopicsList());
+ patternConsumerUpdateQueue.appendTopicsAddedOp(update.getNewTopicsList());
}
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueueTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueueTest.java
new file mode 100644
index 00000000000..01f0be6a85e
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueueTest.java
@@ -0,0 +1,247 @@
+/*
+ * 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.client.impl;
+
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import io.netty.util.HashedWheelTimer;
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.AllArgsConstructor;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+@Test(groups = "utils")
+public class PatternConsumerUpdateQueueTest {
+
+ private QueueInstance createInstance(CompletableFuture<Void> customizedRecheckFuture,
+ CompletableFuture<Void> customizedPartialUpdateFuture,
+ CompletableFuture<Void> customizedConsumerInitFuture) {
+ return createInstance(customizedRecheckFuture, customizedPartialUpdateFuture, customizedConsumerInitFuture,
+ null, null);
+ }
+
+ private QueueInstance createInstance(CompletableFuture<Void> customizedRecheckFuture,
+ CompletableFuture<Void> customizedPartialUpdateFuture,
+ CompletableFuture<Void> customizedConsumerInitFuture,
+ Collection<String> successTopics,
+ Collection<String> errorTopics) {
+ HashedWheelTimer timer = new HashedWheelTimer(new ExecutorProvider.ExtendedThreadFactory("timer-x",
+ Thread.currentThread().isDaemon()), 1, TimeUnit.MILLISECONDS);
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ when(client.timer()).thenReturn(timer);
+
+ PatternMultiTopicsConsumerImpl patternConsumer = mock(PatternMultiTopicsConsumerImpl.class);
+ when(patternConsumer.recheckTopicsChange()).thenReturn(customizedRecheckFuture);
+ when(patternConsumer.getClient()).thenReturn(client);
+ if (customizedConsumerInitFuture != null) {
+ when(patternConsumer.getSubscribeFuture()).thenReturn(customizedConsumerInitFuture);
+ } else {
+ when(patternConsumer.getSubscribeFuture()).thenReturn(CompletableFuture.completedFuture(null));
+ }
+
+ PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener =
+ mock(PatternMultiTopicsConsumerImpl.TopicsChangedListener.class);
+ if (successTopics == null && errorTopics == null) {
+ when(topicsChangeListener.onTopicsAdded(anyCollection())).thenReturn(customizedPartialUpdateFuture);
+ when(topicsChangeListener.onTopicsRemoved(anyCollection())).thenReturn(customizedPartialUpdateFuture);
+ } else {
+ CompletableFuture<Void> ex = FutureUtil.failedFuture(new RuntimeException("mock error"));
+ when(topicsChangeListener.onTopicsAdded(successTopics)).thenReturn(customizedPartialUpdateFuture);
+ when(topicsChangeListener.onTopicsRemoved(successTopics)).thenReturn(customizedPartialUpdateFuture);
+ when(topicsChangeListener.onTopicsAdded(errorTopics)).thenReturn(ex);
+ when(topicsChangeListener.onTopicsRemoved(errorTopics)).thenReturn(ex);
+ }
+
+ PatternConsumerUpdateQueue queue = new PatternConsumerUpdateQueue(patternConsumer, topicsChangeListener);
+ return new QueueInstance(queue, patternConsumer, topicsChangeListener);
+ }
+
+ private QueueInstance createInstance() {
+ CompletableFuture<Void> completedFuture = CompletableFuture.completedFuture(null);
+ return createInstance(completedFuture, completedFuture, completedFuture);
+ }
+
+ @AllArgsConstructor
+ private static class QueueInstance implements Closeable {
+ private PatternConsumerUpdateQueue queue;
+ private PatternMultiTopicsConsumerImpl mockedConsumer;
+ private PatternMultiTopicsConsumerImpl.TopicsChangedListener mockedListener;
+
+ @Override
+ public void close() {
+ mockedConsumer.getClient().timer().stop();
+ }
+ }
+
+ @Test
+ public void testTopicsChangedEvents() {
+ QueueInstance instance = createInstance();
+
+ Collection<String> topics = Arrays.asList("a");
+ for (int i = 0; i < 10; i++) {
+ instance.queue.appendTopicsAddedOp(topics);
+ instance.queue.appendTopicsRemovedOp(topics);
+ }
+ Awaitility.await().untilAsserted(() -> {
+ verify(instance.mockedListener, times(10)).onTopicsAdded(topics);
+ verify(instance.mockedListener, times(10)).onTopicsRemoved(topics);
+ });
+
+ // cleanup.
+ instance.close();
+ }
+
+ @Test
+ public void testRecheckTask() {
+ QueueInstance instance = createInstance();
+
+ for (int i = 0; i < 10; i++) {
+ instance.queue.appendRecheckOp();
+ }
+
+ Awaitility.await().untilAsserted(() -> {
+ verify(instance.mockedConsumer, times(10)).recheckTopicsChange();
+ });
+
+ // cleanup.
+ instance.close();
+ }
+
+ @Test
+ public void testDelayedRecheckTask() {
+ CompletableFuture<Void> recheckFuture = new CompletableFuture<>();
+ CompletableFuture<Void> partialUpdateFuture = CompletableFuture.completedFuture(null);
+ CompletableFuture<Void> consumerInitFuture = CompletableFuture.completedFuture(null);
+ QueueInstance instance = createInstance(recheckFuture, partialUpdateFuture, consumerInitFuture);
+
+ for (int i = 0; i < 10; i++) {
+ instance.queue.appendRecheckOp();
+ }
+
+ recheckFuture.complete(null);
+ Awaitility.await().untilAsserted(() -> {
+ // The first task will be running, and never completed until all tasks have been added.
+ // Since the first was started, the second one will not be skipped.
+ // The others after the second task will be skipped.
+ // So the times that called "recheckTopicsChange" will be 2.
+ verify(instance.mockedConsumer, times(2)).recheckTopicsChange();
+ });
+
+ // cleanup.
+ instance.close();
+ }
+
+ @Test
+ public void testCompositeTasks() {
+ CompletableFuture<Void> recheckFuture = new CompletableFuture<>();
+ CompletableFuture<Void> partialUpdateFuture = CompletableFuture.completedFuture(null);
+ CompletableFuture<Void> consumerInitFuture = CompletableFuture.completedFuture(null);
+ QueueInstance instance = createInstance(recheckFuture, partialUpdateFuture, consumerInitFuture);
+
+ Collection<String> topics = Arrays.asList("a");
+ for (int i = 0; i < 10; i++) {
+ instance.queue.appendRecheckOp();
+ instance.queue.appendTopicsAddedOp(topics);
+ instance.queue.appendTopicsRemovedOp(topics);
+ }
+ recheckFuture.complete(null);
+ Awaitility.await().untilAsserted(() -> {
+ // The first task will be running, and never completed until all tasks have been added.
+ // Since the first was started, the second one will not be skipped.
+ // The others after the second task will be skipped.
+ // So the times that called "recheckTopicsChange" will be 2.
+ verify(instance.mockedConsumer, times(2)).recheckTopicsChange();
+ // The tasks after the second "recheckTopicsChange" will be skipped due to there is a previous
+ // "recheckTopicsChange" that has not been executed.
+ // The tasks between the fist "recheckTopicsChange" and the second "recheckTopicsChange" will be skipped
+ // due to there is a following "recheckTopicsChange".
+ verify(instance.mockedListener, times(0)).onTopicsAdded(topics);
+ verify(instance.mockedListener, times(0)).onTopicsRemoved(topics);
+ });
+
+ // cleanup.
+ instance.close();
+ }
+
+ @Test
+ public void testErrorTask() {
+ CompletableFuture<Void> immediatelyCompleteFuture = CompletableFuture.completedFuture(null);
+ Collection<String> successTopics = Arrays.asList("a");
+ Collection<String> errorTopics = Arrays.asList(UUID.randomUUID().toString());
+ QueueInstance instance = createInstance(immediatelyCompleteFuture, immediatelyCompleteFuture,
+ immediatelyCompleteFuture, successTopics, errorTopics);
+
+ instance.queue.appendTopicsAddedOp(successTopics);
+ instance.queue.appendTopicsRemovedOp(successTopics);
+ instance.queue.appendTopicsAddedOp(errorTopics);
+ instance.queue.appendTopicsAddedOp(successTopics);
+ instance.queue.appendTopicsRemovedOp(successTopics);
+
+ Awaitility.await().atMost(Duration.ofSeconds(60)).untilAsserted(() -> {
+ verify(instance.mockedListener, times(2)).onTopicsAdded(successTopics);
+ verify(instance.mockedListener, times(2)).onTopicsRemoved(successTopics);
+ verify(instance.mockedListener, times(1)).onTopicsAdded(errorTopics);
+ // After an error task will push a recheck task to offset.
+ verify(instance.mockedConsumer, times(1)).recheckTopicsChange();
+ });
+
+ // cleanup.
+ instance.close();
+ }
+
+ @Test
+ public void testFailedSubscribe() {
+ CompletableFuture<Void> immediatelyCompleteFuture = CompletableFuture.completedFuture(null);
+ CompletableFuture<Void> consumerInitFuture = new CompletableFuture<>();
+ Collection<String> successTopics = Arrays.asList("a");
+ Collection<String> errorTopics = Arrays.asList(UUID.randomUUID().toString());
+ QueueInstance instance = createInstance(immediatelyCompleteFuture, immediatelyCompleteFuture,
+ consumerInitFuture, successTopics, errorTopics);
+
+ instance.queue.appendTopicsAddedOp(successTopics);
+ instance.queue.appendTopicsRemovedOp(successTopics);
+ instance.queue.appendTopicsAddedOp(errorTopics);
+ instance.queue.appendTopicsAddedOp(successTopics);
+ instance.queue.appendTopicsRemovedOp(successTopics);
+
+ // Consumer init failed after multi topics changes.
+ // All the topics changes events should be skipped.
+ consumerInitFuture.completeExceptionally(new RuntimeException("mocked ex"));
+ Awaitility.await().untilAsserted(() -> {
+ verify(instance.mockedListener, times(0)).onTopicsAdded(successTopics);
+ verify(instance.mockedListener, times(0)).onTopicsRemoved(successTopics);
+ verify(instance.mockedListener, times(0)).onTopicsAdded(errorTopics);
+ verify(instance.mockedConsumer, times(0)).recheckTopicsChange();
+ });
+
+ // cleanup.
+ instance.close();
+ }
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImplTest.java
index 5baca24cf8a..9c239785da7 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImplTest.java
@@ -61,7 +61,7 @@ public class PatternMultiTopicsConsumerImplTest {
"persistent://tenant/my-ns/non-matching"),
null, false, true),
mockListener,
- Collections.emptyList());
+ Collections.emptyList(), "");
verify(mockListener).onTopicsAdded(Sets.newHashSet(
"persistent://tenant/my-ns/name-1",
"persistent://tenant/my-ns/name-2"));
@@ -80,7 +80,7 @@ public class PatternMultiTopicsConsumerImplTest {
"persistent://tenant/my-ns/name-2"),
"TOPICS_HASH", true, true),
mockListener,
- Arrays.asList("persistent://tenant/my-ns/name-0"));
+ Arrays.asList("persistent://tenant/my-ns/name-0"), "");
verify(mockListener).onTopicsAdded(Sets.newHashSet(
"persistent://tenant/my-ns/name-1",
"persistent://tenant/my-ns/name-2"));
@@ -99,7 +99,7 @@ public class PatternMultiTopicsConsumerImplTest {
"persistent://tenant/my-ns/name-2"),
"TOPICS_HASH", true, false),
mockListener,
- Arrays.asList("persistent://tenant/my-ns/name-0"));
+ Arrays.asList("persistent://tenant/my-ns/name-0"), "");
verify(mockListener, never()).onTopicsAdded(any());
verify(mockListener, never()).onTopicsRemoved(any());
verify(mockTopicsHashSetter).accept("TOPICS_HASH");
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java
index 7e9fd601d4f..1d1a17d50eb 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java
@@ -29,6 +29,7 @@ import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.naming.NamespaceName;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
@@ -68,8 +69,17 @@ public class TopicListWatcherTest {
thenReturn(clientCnxFuture.thenApply(clientCnx -> Pair.of(clientCnx, false)));
when(client.getConnection(any(), any(), anyInt())).thenReturn(clientCnxFuture);
when(connectionPool.getConnection(any(), any(), anyInt())).thenReturn(clientCnxFuture);
+
+ CompletableFuture<Void> completedFuture = CompletableFuture.completedFuture(null);
+ PatternMultiTopicsConsumerImpl patternConsumer = mock(PatternMultiTopicsConsumerImpl.class);
+ when(patternConsumer.getSubscribeFuture()).thenReturn(completedFuture);
+ when(patternConsumer.recheckTopicsChange()).thenReturn(completedFuture);
+ when(listener.onTopicsAdded(anyCollection())).thenReturn(completedFuture);
+ when(listener.onTopicsRemoved(anyCollection())).thenReturn(completedFuture);
+ PatternConsumerUpdateQueue queue = new PatternConsumerUpdateQueue(patternConsumer, listener);
+
watcherFuture = new CompletableFuture<>();
- watcher = new TopicListWatcher(listener, client,
+ watcher = new TopicListWatcher(queue, client,
Pattern.compile(topic), 7,
NamespaceName.get("tenant/ns"), null, watcherFuture, () -> {});
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/lookup/GetTopicsResult.java b/pulsar-common/src/main/java/org/apache/pulsar/common/lookup/GetTopicsResult.java
index 80f16e6c367..d36595d91ef 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/lookup/GetTopicsResult.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/lookup/GetTopicsResult.java
@@ -20,7 +20,10 @@ package org.apache.pulsar.common.lookup;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
+import java.util.regex.Pattern;
import lombok.Getter;
import lombok.ToString;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
@@ -119,4 +122,25 @@ public class GetTopicsResult {
return topics;
}
}
+
+ public GetTopicsResult filterTopics(Pattern topicsPattern) {
+ List<String> topicsFiltered = TopicList.filterTopics(getTopics(), topicsPattern);
+ // If nothing changed.
+ if (topicsFiltered.equals(getTopics())) {
+ GetTopicsResult newObj = new GetTopicsResult(nonPartitionedOrPartitionTopics, null, true, true);
+ newObj.topics = topics;
+ return newObj;
+ }
+ // Filtered some topics.
+ Set<String> topicsFilteredSet = new HashSet<>(topicsFiltered);
+ List<String> newTps = new ArrayList<>();
+ for (String tp: nonPartitionedOrPartitionTopics) {
+ if (topicsFilteredSet.contains(TopicName.get(tp).getPartitionedTopicName())) {
+ newTps.add(tp);
+ }
+ }
+ GetTopicsResult newObj = new GetTopicsResult(newTps, null, true, true);
+ newObj.topics = topicsFiltered;
+ return newObj;
+ }
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index e051e01495d..2298a8d06a8 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.common.util.Codec;
@@ -102,6 +103,14 @@ public class TopicName implements ServiceUnitId {
}
}
+ public static String getPartitionPattern(String topic) {
+ return "^" + Pattern.quote(get(topic).getPartitionedTopicName().toString()) + "-partition-[0-9]+$";
+ }
+
+ public static String getPattern(String topic) {
+ return "^" + Pattern.quote(get(topic).getPartitionedTopicName().toString()) + "$";
+ }
+
@SuppressFBWarnings("DCN_NULLPOINTER_EXCEPTION")
private TopicName(String completeTopicName) {
try {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java b/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java
index 4c0a8d500b7..a66d327dbab 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java
@@ -28,7 +28,9 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import lombok.experimental.UtilityClass;
import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.TopicType;
@UtilityClass
public class TopicList {
@@ -86,9 +88,18 @@ public class TopicList {
if (!originalRegexp.toString().contains(SCHEME_SEPARATOR)) {
return originalRegexp;
}
- String removedTopicDomain = SCHEME_SEPARATOR_PATTERN.split(originalRegexp.toString())[1];
+ String[] parts = SCHEME_SEPARATOR_PATTERN.split(originalRegexp.toString());
+ String prefix = parts[0];
+ String removedTopicDomain = parts[1];
+ if (prefix.equals(TopicDomain.persistent.value()) || prefix.equals(TopicDomain.non_persistent.value())) {
+ prefix = "";
+ } else if (prefix.endsWith(TopicDomain.persistent.value())){
+ prefix = prefix.substring(0, prefix.length() - TopicDomain.persistent.value().length());
+ } else if (prefix.endsWith(TopicDomain.non_persistent.value())) {
+ prefix = prefix.substring(0, prefix.length() - TopicDomain.non_persistent.value().length());
+ }
if (originalRegexp.contains("^")) {
- return String.format("^%s", removedTopicDomain);
+ return String.format("%s%s", prefix, removedTopicDomain);
} else {
return removedTopicDomain;
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
index 0628d494af3..454eee0f966 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
@@ -54,6 +54,9 @@ public class FutureUtil {
* @return a new CompletableFuture that is completed when all of the given CompletableFutures complete
*/
public static CompletableFuture<Void> waitForAll(Collection<? extends CompletableFuture<?>> futures) {
+ if (futures == null || futures.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
}