From b375e869accc5bb4588974025d04edbb0e9911eb Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 19 Feb 2024 16:43:39 +0800 Subject: [PATCH] [fix] [broker] Fix can not subscribe partitioned topic with a suffix-matched regexp (#22025) --- .../broker/resources/TopicResources.java | 3 ++ .../broker/namespace/NamespaceService.java | 3 ++ .../service/PulsarCommandSenderImpl.java | 6 ++++ .../broker/service/TopicListService.java | 22 ++++++++++-- .../impl/PatternTopicsConsumerImplTest.java | 34 +++++++++++++++---- .../pulsar/client/api/ConsumerBuilder.java | 8 ++--- .../client/impl/MultiTopicsConsumerImpl.java | 10 ++++-- .../impl/PatternMultiTopicsConsumerImpl.java | 24 +++++++++++-- .../pulsar/client/impl/TopicListWatcher.java | 4 ++- .../impl/conf/ConsumerConfigurationData.java | 2 +- .../pulsar/common/protocol/Commands.java | 7 ++++ 11 files changed, 104 insertions(+), 19 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 840ced0a1c1c4..0963f25c3d31f 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 TopicResources(MetadataStore store) { store.registerListener(this::handleNotification); } + /*** + * List persistent topics names under a namespace, the topic name contains the partition suffix. + */ public CompletableFuture> 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 d8c3fd169a205..b55eda150afe6 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 @@ -1432,6 +1432,9 @@ private CompletableFuture> getPartitionsForTopic(TopicName topicNam }); } + /*** + * List persistent topics names under a namespace, the topic name contains the partition suffix. + */ public CompletableFuture> 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 dd74fc4e71ed2..105650caaaf13 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 @@ -356,12 +356,18 @@ public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError err writeAndFlush(outBuf); } + /*** + * @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 topics) { BaseCommand command = Commands.newWatchTopicListSuccess(requestId, watcherId, topicsHash, topics); interceptAndWriteCommand(command); } + /*** + * {@inheritDoc} + */ @Override public void sendWatchTopicListUpdate(long watcherId, List newTopics, List 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 7aa50057d73c9..aea5b9fc65b46 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.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 { + /** Topic names which are matching, the topic name contains the partition suffix. **/ private final List 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 topics) { this.topicListService = topicListService; @@ -59,9 +65,12 @@ public List getMatchingTopics() { 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 newTopics; List deletedTopics; if (notificationType == NotificationType.Deleted) { @@ -109,6 +118,9 @@ public void inactivate() { } } + /*** + * @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 void handleWatchTopicList(NamespaceName namespaceName, long watcherId, lo }); } - + /*** + * @param topicsPattern The regexp for the topic name(not contains partition suffix). + */ public void initializeTopicsListWatcher(CompletableFuture watcherFuture, NamespaceName namespace, long watcherId, Pattern topicsPattern) { namespaceService.getListOfPersistentTopics(namespace). @@ -246,6 +260,10 @@ public void deleteTopicListWatcher(Long watcherId) { 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 deletedTopics, List 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 9bcbdfed4c9ee..7707abafde8de 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 @@ -681,16 +681,27 @@ public void testAutoSubscribePatterConsumerFromBrokerWatcher(boolean delayWatchi } } - @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)); @@ -704,6 +715,9 @@ public void testPreciseRegexpSubscribe(boolean partitioned) throws Exception { .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS) .receiverQueueSize(4) .subscribe(); + if (createTopicAfterWatcherStarted) { + waitForTopicListWatcherStarted(consumer); + } // 1. create topic. if (partitioned) { @@ -733,6 +747,14 @@ public void testPreciseRegexpSubscribe(boolean partitioned) throws Exception { } } + @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 6a9aee63fce6c..863432b478fb2 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 extends Cloneable { ConsumerBuilder topics(List topicNames); /** - * Specify a pattern for topics that this consumer subscribes to. + * Specify a pattern for topics(not contains the partition suffix) that this consumer subscribes to. * *

The pattern is applied to subscribe to all topics, within a single namespace, that match the * pattern. @@ -134,13 +134,13 @@ public interface ConsumerBuilder extends Cloneable { *

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 topicsPattern(Pattern topicsPattern); /** - * Specify a pattern for topics that this consumer subscribes to. + * Specify a pattern for topics(not contains the partition suffix) that this consumer subscribes to. * *

It accepts a regular expression that is compiled into a pattern internally. E.g., * "persistent://public/default/pattern-topic-.*" @@ -151,7 +151,7 @@ public interface ConsumerBuilder extends Cloneable { *

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 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 84504b632add2..d18af475d6167 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 @@ -929,7 +929,10 @@ private void removeTopic(String topic) { } } - // subscribe one more given topic + /*** + * Subscribe one more given topic. + * @param topicName topic name without the partition suffix. + */ public CompletableFuture subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) { TopicName topicNameInstance = getTopicName(topicName); if (topicNameInstance == null) { @@ -1251,7 +1254,10 @@ public CompletableFuture unsubscribeAsync(String topicName) { return unsubscribeFuture; } - // Remove a consumer for a topic + /*** + * Remove a consumer for a topic. + * @param topicName topic name contains the partition suffix. + */ public CompletableFuture 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 f3ebcdee6c0d9..4d179f7d914c2 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 @@ -67,6 +67,9 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl onTopicsRemoved(Collection 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 onTopicsAdded(Collection addedTopics); } private class PatternTopicsChangedListener implements TopicsChangedListener { + + /** + * {@inheritDoc} + */ @Override public CompletableFuture onTopicsRemoved(Collection removedTopics) { CompletableFuture removeFuture = new CompletableFuture<>(); @@ -249,6 +264,9 @@ public CompletableFuture onTopicsRemoved(Collection removedTopics) return removeFuture; } + /** + * {@inheritDoc} + */ @Override public CompletableFuture onTopicsAdded(Collection addedTopics) { CompletableFuture 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 489a07a606eb2..86adf69f06e0f 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 @@ -59,6 +59,9 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler. private final Runnable recheckTopicsChangeAfterReconnect; + /*** + * @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, @@ -142,7 +145,6 @@ public CompletableFuture connectionOpened(ClientCnx cnx) { return; } } - this.connectionHandler.resetBackoff(); recheckTopicsChangeAfterReconnect.run(); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java index 8760926792cd7..3ae0e977d13c4 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java @@ -65,7 +65,7 @@ public class ConsumerConfigurationData implements Serializable, Cloneable { @ApiModelProperty( name = "topicsPattern", - value = "Topic pattern" + value = "The regexp for the topic name(not contains partition suffix)." ) private Pattern topicsPattern; 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 34d47e2836bb2..65674af0ae14e 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 @@ -1585,6 +1585,9 @@ public static BaseCommand newWatchTopicList( 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 topics) { BaseCommand cmd = localCmd(Type.WATCH_TOPIC_LIST_SUCCESS); @@ -1600,6 +1603,10 @@ public static BaseCommand newWatchTopicListSuccess(long requestId, long watcherI 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 newTopics, List deletedTopics, String topicsHash) { BaseCommand cmd = localCmd(Type.WATCH_TOPIC_UPDATE);