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:36 UTC
(pulsar) branch branch-2.11 updated (7dafa231aa8 -> 89e061c8652)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 7dafa231aa8 [improve][build] Upgrade Apache ZooKeeper to 3.9.1 (#20933)
new 6af01e5ab09 [fix] [broker] Fix break change: could not subscribe partitioned topic with a suffix-matched regexp due to a mistake of PIP-145 (#21885)
new 89e061c8652 [fix] [broker] Fix can not subscribe partitioned topic with a suffix-matched regexp (#22025)
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:
.../pulsar/broker/resources/TopicResources.java | 3 +
.../pulsar/broker/namespace/NamespaceService.java | 3 +
.../broker/service/PulsarCommandSenderImpl.java | 6 +
.../pulsar/broker/service/TopicListService.java | 22 +++-
.../broker/auth/MockedPulsarServiceBaseTest.java | 14 +++
.../client/impl/PatternTopicsConsumerImplTest.java | 140 ++++++++++++++++++++-
.../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 ++
.../org/apache/pulsar/common/topics/TopicList.java | 21 +++-
12 files changed, 245 insertions(+), 17 deletions(-)
(pulsar) 02/02: [fix] [broker] Fix can not subscribe partitioned topic with a suffix-matched regexp (#22025)
Posted by yu...@apache.org.
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);
(pulsar) 01/02: [fix] [broker] Fix break change: could not subscribe partitioned topic with a suffix-matched regexp due to a mistake of PIP-145 (#21885)
Posted by yu...@apache.org.
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 6af01e5ab0985004479e101d16116b6fd8e232a5
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Jan 15 20:41:35 2024 +0800
[fix] [broker] Fix break change: could not subscribe partitioned topic with a suffix-matched regexp due to a mistake of PIP-145 (#21885)
(cherry picked from commit 4ebbd2f5244ea2f8c0fd75e4dcb52055568b7fc7)
---
.../broker/auth/MockedPulsarServiceBaseTest.java | 14 +++
.../client/impl/PatternTopicsConsumerImplTest.java | 108 +++++++++++++++++++++
.../org/apache/pulsar/common/topics/TopicList.java | 21 +++-
3 files changed, 140 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index 000ad97e490..f298a58b764 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -30,6 +30,7 @@ import com.google.common.collect.Sets;
import com.google.common.util.concurrent.MoreExecutors;
import io.netty.channel.EventLoopGroup;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URL;
@@ -72,6 +73,8 @@ import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.ConnectionPool;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.client.impl.ProducerImpl;
@@ -712,5 +715,16 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
private PersistentTopic persistentTopic;
}
+ public static void reconnectAllConnections(PulsarClientImpl c) throws Exception {
+ ConnectionPool pool = c.getCnxPool();
+ Method closeAllConnections = ConnectionPool.class.getDeclaredMethod("closeAllConnections", new Class[]{});
+ closeAllConnections.setAccessible(true);
+ closeAllConnections.invoke(pool, new Object[]{});
+ }
+
+ public void reconnectAllConnections() throws Exception {
+ reconnectAllConnections((PulsarClientImpl) pulsarClient);
+ }
+
private static final Logger log = LoggerFactory.getLogger(MockedPulsarServiceBaseTest.class);
}
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 09efd847691..f2436d96d39 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
@@ -26,6 +26,7 @@ import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
+import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -33,6 +34,7 @@ import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.IntStream;
+import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
@@ -50,6 +52,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker-impl")
@@ -597,6 +600,111 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
consumer.close();
}
+ @DataProvider(name= "partitioned")
+ public Object[][] partitioned(){
+ return new Object[][]{
+ {true},
+ {false}
+ };
+ }
+
+ @Test(timeOut = testTimeout, dataProvider = "partitioned")
+ public void testPreciseRegexpSubscribe(boolean partitioned) 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)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
+ .receiverQueueSize(4)
+ .subscribe();
+
+ // 1. create topic.
+ if (partitioned) {
+ admin.topics().createPartitionedTopic(topicName, 1);
+ } else {
+ admin.topics().createNonPartitionedTopic(topicName);
+ }
+
+ // 2. verify consumer can subscribe the topic.
+ assertSame(pattern, ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern());
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitions().size(), 1);
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getConsumers().size(), 1);
+ if (partitioned) {
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 1);
+ } else {
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 0);
+ }
+ });
+
+ // cleanup.
+ consumer.close();
+ if (partitioned) {
+ admin.topics().deletePartitionedTopic(topicName);
+ } else {
+ admin.topics().delete(topicName);
+ }
+ }
+
+ @Test(timeOut = 240 * 1000, dataProvider = "partitioned")
+ public void testPreciseRegexpSubscribeDisabledTopicWatcher(boolean partitioned) throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String subscriptionName = "s1";
+ final Pattern pattern = Pattern.compile(String.format("%s$", topicName));
+
+ // Close all ServerCnx by close client-side sockets to make the config changes effect.
+ pulsar.getConfig().setEnableBrokerSideSubscriptionPatternEvaluation(false);
+ reconnectAllConnections();
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topicsPattern(pattern)
+ // Disable brokerSideSubscriptionPatternEvaluation will leading disable topic list watcher.
+ // So set patternAutoDiscoveryPeriod to a little value.
+ .patternAutoDiscoveryPeriod(1)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
+ .receiverQueueSize(4)
+ .subscribe();
+
+ // 1. create topic.
+ if (partitioned) {
+ admin.topics().createPartitionedTopic(topicName, 1);
+ } else {
+ admin.topics().createNonPartitionedTopic(topicName);
+ }
+
+ // 2. verify consumer can subscribe the topic.
+ // Since the minimum value of `patternAutoDiscoveryPeriod` is 60s, we set the test timeout to a triple value.
+ assertSame(pattern, ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern());
+ Awaitility.await().atMost(Duration.ofMinutes(3)).untilAsserted(() -> {
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitions().size(), 1);
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getConsumers().size(), 1);
+ if (partitioned) {
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 1);
+ } else {
+ assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 0);
+ }
+ });
+
+ // cleanup.
+ consumer.close();
+ if (partitioned) {
+ admin.topics().deletePartitionedTopic(topicName);
+ } else {
+ admin.topics().delete(topicName);
+ }
+ // Close all ServerCnx by close client-side sockets to make the config changes effect.
+ pulsar.getConfig().setEnableBrokerSideSubscriptionPatternEvaluation(true);
+ reconnectAllConnections();
+ }
+
// simulate subscribe a pattern which has 3 topics, but then matched topic added in.
@Test(timeOut = testTimeout)
public void testAutoSubscribePatternConsumer() throws Exception {
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 0bebd93747b..b231f21e598 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
@@ -47,13 +47,16 @@ public class TopicList {
}
public static List<String> filterTopics(List<String> original, Pattern topicsPattern) {
- final Pattern shortenedTopicsPattern = topicsPattern.toString().contains(SCHEME_SEPARATOR)
- ? Pattern.compile(SCHEME_SEPARATOR_PATTERN.split(topicsPattern.toString())[1]) : topicsPattern;
+ final Pattern shortenedTopicsPattern = Pattern.compile(removeTopicDomainScheme(topicsPattern.toString()));
return original.stream()
.map(TopicName::get)
+ .filter(topicName -> {
+ String partitionedTopicName = topicName.getPartitionedTopicName();
+ String removedScheme = SCHEME_SEPARATOR_PATTERN.split(partitionedTopicName)[1];
+ return shortenedTopicsPattern.matcher(removedScheme).matches();
+ })
.map(TopicName::toString)
- .filter(topic -> shortenedTopicsPattern.matcher(SCHEME_SEPARATOR_PATTERN.split(topic)[1]).matches())
.collect(Collectors.toList());
}
@@ -78,4 +81,16 @@ public class TopicList {
s1.removeAll(list2);
return s1;
}
+
+ private static String removeTopicDomainScheme(String originalRegexp) {
+ if (!originalRegexp.toString().contains(SCHEME_SEPARATOR)) {
+ return originalRegexp;
+ }
+ String removedTopicDomain = SCHEME_SEPARATOR_PATTERN.split(originalRegexp.toString())[1];
+ if (originalRegexp.contains("^")) {
+ return String.format("^%s", removedTopicDomain);
+ } else {
+ return removedTopicDomain;
+ }
+ }
}