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:14 UTC
(pulsar) branch branch-3.3 updated (97a7b70d731 -> 40abd737c81)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 97a7b70d731 [improve] [broker] Trigger offload on topic load (#22652)
new 2b4788d6fe0 [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted (#22854)
new 40abd737c81 [fix][broker]Fix lookupService.getTopicsUnderNamespace can not work with a quote pattern (#23014)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/pulsar/broker/service/TopicGCTest.java | 291 ++++++++++++++++++++-
.../client/api/PatternMultiTopicsConsumerTest.java | 37 +++
.../client/impl/PatternTopicsConsumerImplTest.java | 69 ++++-
.../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 | 20 +-
.../org/apache/pulsar/common/util/FutureUtil.java | 3 +
.../apache/pulsar/common/topics/TopicListTest.java | 58 +++-
19 files changed, 1315 insertions(+), 203 deletions(-)
create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueue.java
create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/PatternConsumerUpdateQueueTest.java
(pulsar) 02/02: [fix][broker]Fix lookupService.getTopicsUnderNamespace can not work with a quote pattern (#23014)
Posted by yu...@apache.org.
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 40abd737c813976cf43c5d30ba7c917947cb0742
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Jul 10 18:08:48 2024 +0800
[fix][broker]Fix lookupService.getTopicsUnderNamespace can not work with a quote pattern (#23014)
(cherry picked from commit 7c0e82739215fbae9e21270d4c70c9a52dd3e403)
---
.../client/impl/PatternTopicsConsumerImplTest.java | 56 +++++++++++++++++++++
.../org/apache/pulsar/common/topics/TopicList.java | 15 +++---
.../apache/pulsar/common/topics/TopicListTest.java | 58 +++++++++++++++++++++-
3 files changed, 120 insertions(+), 9 deletions(-)
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 3bfecae23a5..a6b080cf6c7 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
@@ -27,6 +27,8 @@ import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -49,6 +51,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
@@ -1114,4 +1117,57 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
assertEquals(pulsar.getBrokerService().getTopicIfExists(baseTopicName + "-1").join(), Optional.empty());
assertTrue(pulsar.getBrokerService().getTopicIfExists(baseTopicName + "-2").join().isPresent());
}
+
+ @Test(dataProvider = "partitioned")
+ public void testPatternQuote(boolean partitioned) throws Exception {
+ final NamespaceName namespace = NamespaceName.get("public/default");
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final PulsarClientImpl client = (PulsarClientImpl) pulsarClient;
+ final LookupService lookup = client.getLookup();
+ List<String> expectedRes = new ArrayList<>();
+ if (partitioned) {
+ admin.topics().createPartitionedTopic(topicName, 2);
+ expectedRes.add(TopicName.get(topicName).getPartition(0).toString());
+ expectedRes.add(TopicName.get(topicName).getPartition(1).toString());
+ Collections.sort(expectedRes);
+ } else {
+ admin.topics().createNonPartitionedTopic(topicName);
+ expectedRes.add(topicName);
+ }
+
+ // Verify 1: "java.util.regex.Pattern.quote".
+ String pattern1 = java.util.regex.Pattern.quote(topicName);
+ List<String> res1 = lookup.getTopicsUnderNamespace(namespace, CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ pattern1, null).join().getNonPartitionedOrPartitionTopics();
+ Collections.sort(res1);
+ assertEquals(res1, expectedRes);
+
+ // Verify 2: "com.google.re2j.Pattern.quote"
+ String pattern2 = com.google.re2j.Pattern.quote(topicName);
+ List<String> res2 = lookup.getTopicsUnderNamespace(namespace, CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ pattern2, null).join().getNonPartitionedOrPartitionTopics();
+ Collections.sort(res2);
+ assertEquals(res2, expectedRes);
+
+ // Verify 3: "java.util.regex.Pattern.quote" & "^$"
+ String pattern3 = "^" + java.util.regex.Pattern.quote(topicName) + "$";
+ List<String> res3 = lookup.getTopicsUnderNamespace(namespace, CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ pattern3, null).join().getNonPartitionedOrPartitionTopics();
+ Collections.sort(res3);
+ assertEquals(res3, expectedRes);
+
+ // Verify 4: "com.google.re2j.Pattern.quote" & "^$"
+ String pattern4 = "^" + com.google.re2j.Pattern.quote(topicName) + "$";
+ List<String> res4 = lookup.getTopicsUnderNamespace(namespace, CommandGetTopicsOfNamespace.Mode.PERSISTENT,
+ pattern4, null).join().getNonPartitionedOrPartitionTopics();
+ Collections.sort(res4);
+ assertEquals(res4, expectedRes);
+
+ // cleanup.
+ if (partitioned) {
+ admin.topics().deletePartitionedTopic(topicName, false);
+ } else {
+ admin.topics().delete(topicName, false);
+ }
+ }
}
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 a66d327dbab..4dd48732225 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.common.topics;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.hash.Hashing;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
@@ -30,7 +31,6 @@ 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 {
@@ -84,7 +84,8 @@ public class TopicList {
return s1;
}
- private static String removeTopicDomainScheme(String originalRegexp) {
+ @VisibleForTesting
+ static String removeTopicDomainScheme(String originalRegexp) {
if (!originalRegexp.toString().contains(SCHEME_SEPARATOR)) {
return originalRegexp;
}
@@ -93,15 +94,13 @@ public class TopicList {
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%s", prefix, removedTopicDomain);
+ } else if (prefix.endsWith(TopicDomain.persistent.value())){
+ prefix = prefix.substring(0, prefix.length() - TopicDomain.persistent.value().length());
} else {
- return removedTopicDomain;
+ throw new IllegalArgumentException("Does not support topic domain: " + prefix);
}
+ return String.format("%s%s", prefix, removedTopicDomain);
}
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java
index 9069dd6dcc7..b3a7536ebff 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/topics/TopicListTest.java
@@ -30,6 +30,7 @@ import java.util.stream.Stream;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
public class TopicListTest {
@@ -107,5 +108,60 @@ public class TopicListTest {
}
-
+ @Test
+ public void testRemoveTopicDomainScheme() {
+ // persistent.
+ final String tpName1 = "persistent://public/default/tp";
+ String res1 = TopicList.removeTopicDomainScheme(tpName1);
+ assertEquals(res1, "public/default/tp");
+
+ // non-persistent
+ final String tpName2 = "non-persistent://public/default/tp";
+ String res2 = TopicList.removeTopicDomainScheme(tpName2);
+ assertEquals(res2, "public/default/tp");
+
+ // without topic domain.
+ final String tpName3 = "public/default/tp";
+ String res3 = TopicList.removeTopicDomainScheme(tpName3);
+ assertEquals(res3, "public/default/tp");
+
+ // persistent & "java.util.regex.Pattern.quote".
+ final String tpName4 = java.util.regex.Pattern.quote(tpName1);
+ String res4 = TopicList.removeTopicDomainScheme(tpName4);
+ assertEquals(res4, java.util.regex.Pattern.quote("public/default/tp"));
+
+ // persistent & "java.util.regex.Pattern.quote" & "^$".
+ final String tpName5 = "^" + java.util.regex.Pattern.quote(tpName1) + "$";
+ String res5 = TopicList.removeTopicDomainScheme(tpName5);
+ assertEquals(res5, "^" + java.util.regex.Pattern.quote("public/default/tp") + "$");
+
+ // persistent & "com.google.re2j.Pattern.quote".
+ final String tpName6 = Pattern.quote(tpName1);
+ String res6 = TopicList.removeTopicDomainScheme(tpName6);
+ assertEquals(res6, Pattern.quote("public/default/tp"));
+
+ // non-persistent & "java.util.regex.Pattern.quote".
+ final String tpName7 = java.util.regex.Pattern.quote(tpName2);
+ String res7 = TopicList.removeTopicDomainScheme(tpName7);
+ assertEquals(res7, java.util.regex.Pattern.quote("public/default/tp"));
+
+ // non-persistent & "com.google.re2j.Pattern.quote".
+ final String tpName8 = Pattern.quote(tpName2);
+ String res8 = TopicList.removeTopicDomainScheme(tpName8);
+ assertEquals(res8, Pattern.quote("public/default/tp"));
+
+ // non-persistent & "com.google.re2j.Pattern.quote" & "^$".
+ final String tpName9 = "^" + Pattern.quote(tpName2) + "$";
+ String res9 = TopicList.removeTopicDomainScheme(tpName9);
+ assertEquals(res9, "^" + Pattern.quote("public/default/tp") + "$");
+
+ // wrong topic domain.
+ final String tpName10 = "xx://public/default/tp";
+ try {
+ TopicList.removeTopicDomainScheme(tpName10);
+ fail("Does not support the topic domain xx");
+ } catch (Exception ex) {
+ // expected error.
+ }
+ }
}
(pulsar) 01/02: [fix][client] Fix pattern consumer create crash if a part of partitions of a topic have been deleted (#22854)
Posted by yu...@apache.org.
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]));
}