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/02/28 07:24:38 UTC
(pulsar) 02/02: [fix] [broker] Fix can not subscribe partitioned topic with a suffix-matched regexp (#22025)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 89e061c86523db338d7fa4e36ef546e9f5246289
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Feb 19 16:43:39 2024 +0800
[fix] [broker] Fix can not subscribe partitioned topic with a suffix-matched regexp (#22025)
(cherry picked from commit b375e869accc5bb4588974025d04edbb0e9911eb)
---
.../pulsar/broker/resources/TopicResources.java | 3 ++
.../pulsar/broker/namespace/NamespaceService.java | 3 ++
.../broker/service/PulsarCommandSenderImpl.java | 6 +++
.../pulsar/broker/service/TopicListService.java | 22 +++++++++-
.../client/impl/PatternTopicsConsumerImplTest.java | 48 +++++++++++++++++-----
.../apache/pulsar/client/api/ConsumerBuilder.java | 8 ++--
.../client/impl/MultiTopicsConsumerImpl.java | 10 ++++-
.../impl/PatternMultiTopicsConsumerImpl.java | 24 +++++++++--
.../pulsar/client/impl/TopicListWatcher.java | 4 +-
.../apache/pulsar/common/protocol/Commands.java | 7 ++++
10 files changed, 113 insertions(+), 22 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java
index ee537431280..ca273f0ddfa 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java
@@ -50,6 +50,9 @@ public class TopicResources {
store.registerListener(this::handleNotification);
}
+ /***
+ * List persistent topics names under a namespace, the topic name contains the partition suffix.
+ */
public CompletableFuture<List<String>> listPersistentTopicsAsync(NamespaceName ns) {
String path = MANAGED_LEDGER_PATH + "/" + ns + "/persistent";
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 12dc31f3428..e9eacf570e3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1254,6 +1254,9 @@ public class NamespaceService implements AutoCloseable {
});
}
+ /***
+ * List persistent topics names under a namespace, the topic name contains the partition suffix.
+ */
public CompletableFuture<List<String>> getListOfPersistentTopics(NamespaceName namespaceName) {
return pulsar.getPulsarResources().getTopicResources().listPersistentTopicsAsync(namespaceName);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java
index 8dd5a451b2b..11b62e5b9f9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java
@@ -363,12 +363,18 @@ public class PulsarCommandSenderImpl implements PulsarCommandSender {
}
}
+ /***
+ * @param topics topic names which are matching, the topic name contains the partition suffix.
+ */
@Override
public void sendWatchTopicListSuccess(long requestId, long watcherId, String topicsHash, List<String> topics) {
BaseCommand command = Commands.newWatchTopicListSuccess(requestId, watcherId, topicsHash, topics);
interceptAndWriteCommand(command);
}
+ /***
+ * {@inheritDoc}
+ */
@Override
public void sendWatchTopicListUpdate(long watcherId,
List<String> newTopics, List<String> deletedTopics, String topicsHash) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java
index 966e6d0fa30..9e13ad58039 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java
@@ -31,6 +31,7 @@ import org.apache.pulsar.broker.resources.TopicResources;
import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.topics.TopicList;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
import org.apache.pulsar.metadata.api.NotificationType;
@@ -42,11 +43,16 @@ public class TopicListService {
public static class TopicListWatcher implements BiConsumer<String, NotificationType> {
+ /** Topic names which are matching, the topic name contains the partition suffix. **/
private final List<String> matchingTopics;
private final TopicListService topicListService;
private final long id;
+ /** The regexp for the topic name(not contains partition suffix). **/
private final Pattern topicsPattern;
+ /***
+ * @param topicsPattern The regexp for the topic name(not contains partition suffix).
+ */
public TopicListWatcher(TopicListService topicListService, long id,
Pattern topicsPattern, List<String> topics) {
this.topicListService = topicListService;
@@ -59,9 +65,12 @@ public class TopicListService {
return matchingTopics;
}
+ /***
+ * @param topicName topic name which contains partition suffix.
+ */
@Override
public void accept(String topicName, NotificationType notificationType) {
- if (topicsPattern.matcher(topicName).matches()) {
+ if (topicsPattern.matcher(TopicName.get(topicName).getPartitionedTopicName()).matches()) {
List<String> newTopics;
List<String> deletedTopics;
if (notificationType == NotificationType.Deleted) {
@@ -109,6 +118,9 @@ public class TopicListService {
}
}
+ /***
+ * @param topicsPattern The regexp for the topic name(not contains partition suffix).
+ */
public void handleWatchTopicList(NamespaceName namespaceName, long watcherId, long requestId, Pattern topicsPattern,
String topicsHash, Semaphore lookupSemaphore) {
@@ -184,7 +196,9 @@ public class TopicListService {
});
}
-
+ /***
+ * @param topicsPattern The regexp for the topic name(not contains partition suffix).
+ */
public void initializeTopicsListWatcher(CompletableFuture<TopicListWatcher> watcherFuture,
NamespaceName namespace, long watcherId, Pattern topicsPattern) {
namespaceService.getListOfPersistentTopics(namespace).
@@ -242,6 +256,10 @@ public class TopicListService {
log.info("[{}] Closed watcher, watcherId={}", connection.getRemoteAddress(), watcherId);
}
+ /**
+ * @param deletedTopics topic names deleted(contains the partition suffix).
+ * @param newTopics topics names added(contains the partition suffix).
+ */
public void sendTopicListUpdate(long watcherId, String topicsHash, List<String> deletedTopics,
List<String> newTopics) {
connection.getCommandSender().sendWatchTopicListUpdate(watcherId, newTopics, deletedTopics, topicsHash);
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 f2436d96d39..6c9ea2a2efc 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
@@ -48,6 +48,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
@@ -68,6 +69,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
isTcpLookup = true;
// enabled transaction, to test pattern consumers not subscribe to transaction system topic.
conf.setTransactionCoordinatorEnabled(true);
+ conf.setSubscriptionPatternMaxLength(10000);
super.internalSetup();
super.producerBaseSetup();
}
@@ -574,8 +576,9 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
Pattern pattern = Pattern.compile("persistent://my-property/my-ns/pattern-topic.*");
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topicsPattern(pattern)
- // Disable automatic discovery.
- .patternAutoDiscoveryPeriod(1000)
+ // Enable automatic discovery.
+ // See https://github.com/apache/pulsar/pull/20779 and https://github.com/apache/pulsar/pull/21853.
+ .patternAutoDiscoveryPeriod(0)
.subscriptionName(subscriptionName)
.subscriptionType(SubscriptionType.Shared)
.ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
@@ -597,32 +600,49 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 1);
});
+ // cleanup.
consumer.close();
+ admin.topics().deletePartitionedTopic(topicName, false);
}
- @DataProvider(name= "partitioned")
- public Object[][] partitioned(){
+ @DataProvider(name= "regexpConsumerArgs")
+ public Object[][] regexpConsumerArgs(){
return new Object[][]{
- {true},
- {false}
+ //{true, true},
+ {true, false},
+ //{false, true},
+ //{false, false}
};
}
- @Test(timeOut = testTimeout, dataProvider = "partitioned")
- public void testPreciseRegexpSubscribe(boolean partitioned) throws Exception {
+ private void waitForTopicListWatcherStarted(Consumer consumer) {
+ Awaitility.await().untilAsserted(() -> {
+ CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
+ log.info("isDone: {}, isCompletedExceptionally: {}", completableFuture.isDone(),
+ completableFuture.isCompletedExceptionally());
+ assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
+ });
+ }
+
+ @Test(timeOut = testTimeout, dataProvider = "regexpConsumerArgs")
+ public void testPreciseRegexpSubscribe(boolean partitioned, boolean createTopicAfterWatcherStarted) throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
final String subscriptionName = "s1";
final Pattern pattern = Pattern.compile(String.format("%s$", topicName));
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topicsPattern(pattern)
- // Disable automatic discovery.
- .patternAutoDiscoveryPeriod(1000)
+ // Enable automatic discovery.
+ // See https://github.com/apache/pulsar/pull/20779 and https://github.com/apache/pulsar/pull/21853.
+ .patternAutoDiscoveryPeriod(0)
.subscriptionName(subscriptionName)
.subscriptionType(SubscriptionType.Shared)
.ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
.receiverQueueSize(4)
.subscribe();
+ if (createTopicAfterWatcherStarted) {
+ waitForTopicListWatcherStarted(consumer);
+ }
// 1. create topic.
if (partitioned) {
@@ -652,6 +672,14 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
}
}
+ @DataProvider(name= "partitioned")
+ public Object[][] partitioned(){
+ return new Object[][]{
+ {true},
+ {true}
+ };
+ }
+
@Test(timeOut = 240 * 1000, dataProvider = "partitioned")
public void testPreciseRegexpSubscribeDisabledTopicWatcher(boolean partitioned) throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
index aa6d860f691..9e9fcd91c6c 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
@@ -126,7 +126,7 @@ public interface ConsumerBuilder<T> extends Cloneable {
ConsumerBuilder<T> topics(List<String> topicNames);
/**
- * Specify a pattern for topics that this consumer subscribes on.
+ * Specify a pattern for topics(not contains the partition suffix) that this consumer subscribes to.
*
* <p>The pattern is applied to subscribe to all the topics, within a single namespace, that matches the
* pattern.
@@ -134,13 +134,13 @@ public interface ConsumerBuilder<T> extends Cloneable {
* <p>The consumer automatically subscribes to topics created after itself.
*
* @param topicsPattern
- * a regular expression to select a list of topics to subscribe to
+ * a regular expression to select a list of topics(not contains the partition suffix) to subscribe to
* @return the consumer builder instance
*/
ConsumerBuilder<T> topicsPattern(Pattern topicsPattern);
/**
- * Specify a pattern for topics that this consumer subscribes on.
+ * Specify a pattern for topics(not contains the partition suffix) that this consumer subscribes to.
*
* <p>It accepts regular expression that is compiled into a pattern internally. Eg.
* "persistent://public/default/pattern-topic-.*"
@@ -151,7 +151,7 @@ public interface ConsumerBuilder<T> extends Cloneable {
* <p>The consumer automatically subscribes to topics created after itself.
*
* @param topicsPattern
- * given regular expression for topics pattern
+ * given regular expression for topics(not contains the partition suffix) pattern
* @return the consumer builder instance
*/
ConsumerBuilder<T> topicsPattern(String topicsPattern);
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 15ae360ebf5..7965c12c940 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
@@ -928,7 +928,10 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
}
}
- // subscribe one more given topic
+ /***
+ * Subscribe one more given topic.
+ * @param topicName topic name without the partition suffix.
+ */
public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) {
TopicName topicNameInstance = getTopicName(topicName);
if (topicNameInstance == null) {
@@ -1249,7 +1252,10 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
return unsubscribeFuture;
}
- // Remove a consumer for a topic
+ /***
+ * 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);
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 4433bba15d5..a87616d2750 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
@@ -55,6 +55,9 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
private volatile Timeout recheckPatternTimeout = null;
private volatile String topicsHash;
+ /***
+ * @param topicsPattern The regexp for the topic name(not contains partition suffix).
+ */
public PatternMultiTopicsConsumerImpl(Pattern topicsPattern,
String topicsHash,
PulsarClientImpl client,
@@ -167,14 +170,26 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
}
interface TopicsChangedListener {
- // unsubscribe and delete ConsumerImpl in the `consumers` map in `MultiTopicsConsumerImpl` based on added topics
+ /***
+ * unsubscribe and delete {@link ConsumerImpl} in the {@link MultiTopicsConsumerImpl#consumers} map in
+ * {@link MultiTopicsConsumerImpl}.
+ * @param removedTopics topic names removed(contains the partition suffix).
+ */
CompletableFuture<Void> onTopicsRemoved(Collection<String> removedTopics);
- // subscribe and create a list of new ConsumerImpl, added them to the `consumers` map in
- // `MultiTopicsConsumerImpl`.
+
+ /***
+ * subscribe and create a list of new {@link ConsumerImpl}, added them to the
+ * {@link MultiTopicsConsumerImpl#consumers} map in {@link MultiTopicsConsumerImpl}.
+ * @param addedTopics topic names added(contains the partition suffix).
+ */
CompletableFuture<Void> onTopicsAdded(Collection<String> addedTopics);
}
private class PatternTopicsChangedListener implements TopicsChangedListener {
+
+ /**
+ * {@inheritDoc}
+ */
@Override
public CompletableFuture<Void> onTopicsRemoved(Collection<String> removedTopics) {
CompletableFuture<Void> removeFuture = new CompletableFuture<>();
@@ -196,6 +211,9 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
return removeFuture;
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public CompletableFuture<Void> onTopicsAdded(Collection<String> addedTopics) {
CompletableFuture<Void> addFuture = new CompletableFuture<>();
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 7b67c195d30..3d3fb337393 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
@@ -57,6 +57,9 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
private final AtomicReference<ClientCnx> clientCnxUsedForWatcherRegistration = new AtomicReference<>();
+ /***
+ * @param topicsPattern The regexp for the topic name(not contains partition suffix).
+ */
public TopicListWatcher(PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener,
PulsarClientImpl client, Pattern topicsPattern, long watcherId,
NamespaceName namespace, String topicsHash,
@@ -138,7 +141,6 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
return;
}
}
-
this.connectionHandler.resetBackoff();
watcherFuture.complete(this);
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index cad3c310908..c2f01553f8d 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -1517,6 +1517,9 @@ public class Commands {
return cmd;
}
+ /***
+ * @param topics topic names which are matching, the topic name contains the partition suffix.
+ */
public static BaseCommand newWatchTopicListSuccess(long requestId, long watcherId, String topicsHash,
List<String> topics) {
BaseCommand cmd = localCmd(Type.WATCH_TOPIC_LIST_SUCCESS);
@@ -1532,6 +1535,10 @@ public class Commands {
return cmd;
}
+ /**
+ * @param deletedTopics topic names deleted(contains the partition suffix).
+ * @param newTopics topics names added(contains the partition suffix).
+ */
public static BaseCommand newWatchTopicUpdate(long watcherId,
List<String> newTopics, List<String> deletedTopics, String topicsHash) {
BaseCommand cmd = localCmd(Type.WATCH_TOPIC_UPDATE);