You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/02/06 08:32:02 UTC
(pulsar) branch master updated: [fix][broker] Prevent missed topic changes in topic watchers and schedule periodic refresh with patternAutoDiscoveryPeriod interval (#25188)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 2e06cc03570 [fix][broker] Prevent missed topic changes in topic watchers and schedule periodic refresh with patternAutoDiscoveryPeriod interval (#25188)
2e06cc03570 is described below
commit 2e06cc03570d69f8a0df0271fd5178438ea06fff
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Feb 6 10:31:56 2026 +0200
[fix][broker] Prevent missed topic changes in topic watchers and schedule periodic refresh with patternAutoDiscoveryPeriod interval (#25188)
---
.../pulsar/broker/resources/PulsarResources.java | 8 +-
.../pulsar/broker/resources/TopicListener.java | 29 ++
.../pulsar/broker/resources/TopicResources.java | 83 ++++-
.../broker/resources/TopicResourcesTest.java | 103 ++++--
.../pulsar/broker/service/TopicListService.java | 375 ++++++++++++++-------
.../pulsar/broker/service/ServerCnxTest.java | 5 +-
.../broker/service/TopicListServiceTest.java | 363 +++++++++++++++++++-
.../broker/service/TopicListWatcherTest.java | 8 +-
.../testcontext/NonStartableTestPulsarService.java | 2 +-
...icWatcherBackPressureMultipleConsumersTest.java | 17 +-
.../impl/AutoCloseUselessClientConSupports.java | 9 +
...eUselessClientConTopicsPatternConsumerTest.java | 3 +
.../client/impl/PatternTopicsConsumerImplTest.java | 195 ++++++-----
.../org/apache/pulsar/client/impl/ClientCnx.java | 15 +
.../client/impl/PatternConsumerUpdateQueue.java | 176 +++++++---
.../impl/PatternMultiTopicsConsumerImpl.java | 216 ++++++++----
.../pulsar/client/impl/PulsarClientImpl.java | 9 +-
.../pulsar/client/impl/TopicListWatcher.java | 75 ++++-
.../pulsar/client/impl/ClientTestFixtures.java | 16 +-
.../impl/PatternConsumerUpdateQueueTest.java | 74 ++--
.../impl/PatternMultiTopicsConsumerImplTest.java | 268 ++++++++++++++-
.../pulsar/client/impl/TopicListWatcherTest.java | 81 ++++-
.../apache/pulsar/common/protocol/Commands.java | 1 +
.../org/apache/pulsar/common/topics/TopicList.java | 2 +-
pulsar-common/src/main/proto/PulsarApi.proto | 5 +-
.../impl/AutoCloseUselessClientConProxyTest.java | 5 +
26 files changed, 1669 insertions(+), 474 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
index cc64eeb52f6..7b89fe69194 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
@@ -52,20 +52,20 @@ public class PulsarResources {
@Getter
private final LoadBalanceResources loadBalanceResources;
@Getter
- private final Optional<MetadataStore> localMetadataStore;
+ private final Optional<MetadataStoreExtended> localMetadataStore;
@Getter
private final Optional<MetadataStore> configurationMetadataStore;
- public PulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore) {
+ public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStore configurationMetadataStore) {
this(localMetadataStore, configurationMetadataStore, DEFAULT_OPERATION_TIMEOUT_SEC);
}
- public PulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore,
+ public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStore configurationMetadataStore,
int operationTimeoutSec) {
this(localMetadataStore, configurationMetadataStore, operationTimeoutSec, ForkJoinPool.commonPool());
}
- public PulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore,
+ public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStore configurationMetadataStore,
int operationTimeoutSec, Executor executor) {
if (configurationMetadataStore != null) {
tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec);
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicListener.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicListener.java
new file mode 100644
index 00000000000..1d862067f15
--- /dev/null
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicListener.java
@@ -0,0 +1,29 @@
+/*
+ * 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.broker.resources;
+
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.metadata.api.NotificationType;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
+
+public interface TopicListener {
+ NamespaceName getNamespaceName();
+ void onTopicEvent(String topicName, NotificationType notificationType);
+ void onSessionEvent(SessionEvent sessionEvent);
+}
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 de7596a7638..91e84096d15 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
@@ -30,24 +30,27 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.common.naming.NamespaceName;
+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.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
@Slf4j
public class TopicResources {
private static final String MANAGED_LEDGER_PATH = "/managed-ledgers";
- private final MetadataStore store;
+ private final MetadataStoreExtended store;
- private final Map<BiConsumer<String, NotificationType>, Pattern> topicListeners;
+ private final Map<TopicListener, Pattern> topicListeners;
- public TopicResources(MetadataStore store) {
+ public TopicResources(MetadataStoreExtended store) {
this.store = store;
topicListeners = new ConcurrentHashMap<>();
store.registerListener(this::handleNotification);
+ store.registerSessionListener(this::handleSessionEvent);
}
/***
@@ -111,13 +114,15 @@ public class TopicResources {
if (notification.getPath().startsWith(MANAGED_LEDGER_PATH)
&& (notification.getType() == NotificationType.Created
|| notification.getType() == NotificationType.Deleted)) {
- for (Map.Entry<BiConsumer<String, NotificationType>, Pattern> entry :
+ for (Map.Entry<TopicListener, Pattern> entry :
topicListeners.entrySet()) {
Matcher matcher = entry.getValue().matcher(notification.getPath());
if (matcher.matches()) {
TopicName topicName = TopicName.get(
matcher.group(2), NamespaceName.get(matcher.group(1)), decode(matcher.group(3)));
- entry.getKey().accept(topicName.toString(), notification.getType());
+ if (!SystemTopicNames.isSystemTopic(topicName)) {
+ entry.getKey().onTopicEvent(topicName.toString(), notification.getType());
+ }
}
}
}
@@ -128,13 +133,75 @@ public class TopicResources {
+ TopicDomain.persistent + ")/(" + "[^/]+)");
}
+ public void registerPersistentTopicListener(TopicListener listener) {
+ topicListeners.put(listener, namespaceNameToTopicNamePattern(listener.getNamespaceName()));
+ }
+
+ public void deregisterPersistentTopicListener(TopicListener listener) {
+ topicListeners.remove(listener);
+ }
+
+ private void handleSessionEvent(SessionEvent sessionEvent) {
+ topicListeners.keySet().forEach(listener -> {
+ try {
+ listener.onSessionEvent(sessionEvent);
+ } catch (Exception e) {
+ log.warn("Failed to handle session event {} for listener {}", sessionEvent, listener, e);
+ }
+ });
+ }
+
+ @Deprecated
public void registerPersistentTopicListener(
NamespaceName namespaceName, BiConsumer<String, NotificationType> listener) {
- topicListeners.put(listener, namespaceNameToTopicNamePattern(namespaceName));
+ topicListeners.put(new BiConsumerTopicListener(listener, namespaceName),
+ namespaceNameToTopicNamePattern(namespaceName));
}
+ @Deprecated
public void deregisterPersistentTopicListener(BiConsumer<String, NotificationType> listener) {
- topicListeners.remove(listener);
+ topicListeners.remove(new BiConsumerTopicListener(listener, null));
}
+ // for backwards compatibility with broker plugins that could be using the BiConsumer based methods
+ @Deprecated
+ static class BiConsumerTopicListener implements TopicListener {
+ private final BiConsumer<String, NotificationType> listener;
+ private final NamespaceName namespaceName;
+
+ BiConsumerTopicListener(BiConsumer<String, NotificationType> listener, NamespaceName namespaceName) {
+ this.listener = listener;
+ this.namespaceName = namespaceName;
+ }
+
+ @Override
+ public NamespaceName getNamespaceName() {
+ return namespaceName;
+ }
+
+ @Override
+ public void onTopicEvent(String topicName, NotificationType notificationType) {
+ listener.accept(topicName, notificationType);
+ }
+
+ @Override
+ public void onSessionEvent(SessionEvent sessionEvent) {
+ // ignore
+ }
+
+ @Override
+ public String toString() {
+ return "BiConsumerTopicListener [listener=" + listener + ", namespaceName=" + namespaceName + "]";
+ }
+
+ @Override
+ public int hashCode() {
+ return listener.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof BiConsumerTopicListener && listener.equals(((BiConsumerTopicListener) obj).listener);
+ }
+ }
}
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java
index 48092e2a830..13c8f8d01c0 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java
@@ -21,24 +21,24 @@ package org.apache.pulsar.broker.resources;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
import java.util.function.BiConsumer;
import org.apache.pulsar.common.naming.NamespaceName;
-import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
public class TopicResourcesTest {
- private MetadataStore metadataStore;
+ private MetadataStoreExtended metadataStore;
private TopicResources topicResources;
@BeforeMethod
public void setup() {
- metadataStore = mock(MetadataStore.class);
+ metadataStore = mock(MetadataStoreExtended.class);
topicResources = new TopicResources(metadataStore);
}
@@ -49,65 +49,87 @@ public class TopicResourcesTest {
@Test
public void testListenerInvokedWhenTopicCreated() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace/persistent/topic"));
- verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Created);
+ verify(listener).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Created);
}
@Test
public void testListenerInvokedWhenTopicV1Created() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/cluster/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/cluster/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/cluster/namespace/persistent/topic"));
- verify(listener).accept("persistent://tenant/cluster/namespace/topic", NotificationType.Created);
+ verify(listener).onTopicEvent("persistent://tenant/cluster/namespace/topic", NotificationType.Created);
}
@Test
public void testListenerInvokedWhenTopicDeleted() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
topicResources.handleNotification(new Notification(NotificationType.Deleted,
"/managed-ledgers/tenant/namespace/persistent/topic"));
- verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Deleted);
+ verify(listener).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Deleted);
}
@Test
public void testListenerNotInvokedWhenSubscriptionCreated() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace/persistent/topic/subscription"));
- verifyNoInteractions(listener);
+ verifyNoMoreInteractions(listener);
}
@Test
public void testListenerNotInvokedWhenTopicCreatedInOtherNamespace() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace2/persistent/topic"));
- verifyNoInteractions(listener);
+ verifyNoMoreInteractions(listener);
}
@Test
public void testListenerNotInvokedWhenTopicModified() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
topicResources.handleNotification(new Notification(NotificationType.Modified,
"/managed-ledgers/tenant/namespace/persistent/topic"));
- verifyNoInteractions(listener);
+ verifyNoMoreInteractions(listener);
+ }
+
+ @Test
+ public void testListenerNotInvokedForSystemTopicChanges() {
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
+ topicResources.handleNotification(new Notification(NotificationType.Created,
+ "/managed-ledgers/tenant/namespace/persistent/__change_events"));
+ verifyNoMoreInteractions(listener);
}
@Test
public void testListenerNotInvokedAfterDeregistered() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace/persistent/topic"));
- verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Created);
+ verify(listener).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Created);
topicResources.deregisterPersistentTopicListener(listener);
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace/persistent/topic2"));
@@ -116,22 +138,39 @@ public class TopicResourcesTest {
@Test
public void testListenerInvokedWithDecodedTopicName() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace/persistent/topic%3Atest"));
- verify(listener).accept("persistent://tenant/namespace/topic:test", NotificationType.Created);
+ verify(listener).onTopicEvent("persistent://tenant/namespace/topic:test", NotificationType.Created);
}
@Test
public void testNamespaceContainsDotsShouldntMatchAny() {
- BiConsumer<String, NotificationType> listener = mock(BiConsumer.class);
- topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/name.pace"), listener);
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/name.pace"));
+ topicResources.registerPersistentTopicListener(listener);
+ verify(listener).getNamespaceName();
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/namespace/persistent/topic"));
- verifyNoInteractions(listener);
+ verifyNoMoreInteractions(listener);
topicResources.handleNotification(new Notification(NotificationType.Created,
"/managed-ledgers/tenant/name.pace/persistent/topic"));
- verify(listener).accept("persistent://tenant/name.pace/topic", NotificationType.Created);
+ verify(listener).onTopicEvent("persistent://tenant/name.pace/topic", NotificationType.Created);
+ }
+
+ @Test
+ public void testBiConsumerListenerNotInvokedAfterDeregistered() {
+ BiConsumer listener = mock(BiConsumer.class);
+ topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener);
+ topicResources.handleNotification(new Notification(NotificationType.Created,
+ "/managed-ledgers/tenant/namespace/persistent/topic"));
+ verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Created);
+ topicResources.deregisterPersistentTopicListener(listener);
+ topicResources.handleNotification(new Notification(NotificationType.Created,
+ "/managed-ledgers/tenant/namespace/persistent/topic2"));
+ verifyNoMoreInteractions(listener);
}
}
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 ec0f9b73e7e..262c734ea95 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
@@ -24,6 +24,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
@@ -35,13 +36,15 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import lombok.Getter;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.resources.TopicListener;
import org.apache.pulsar.broker.resources.TopicResources;
import org.apache.pulsar.broker.topiclistlimit.TopicListMemoryLimiter;
import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache;
@@ -59,11 +62,12 @@ import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
import org.apache.pulsar.metadata.api.NotificationType;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TopicListService {
- public static class TopicListWatcher implements BiConsumer<String, NotificationType> {
+ public static class TopicListWatcher implements TopicListener {
// upper bound for buffered topic list updates
private static final int DEFAULT_TOPIC_LIST_UPDATE_MAX_QUEUE_SIZE = 10000;
/** Topic names which are matching, the topic name contains the partition suffix. **/
@@ -77,7 +81,12 @@ public class TopicListService {
private volatile boolean closed = false;
private boolean sendingInProgress;
private final BlockingDeque<Runnable> sendTopicListUpdateTasks;
+ @Getter(onMethod_ = @VisibleForTesting)
private boolean updatingTopics;
+ private List<String> matchingTopicsBeforeDisconnected;
+ private boolean disconnected;
+ private List<Runnable> updateCallbacks = new LinkedList<>();
+ private boolean updatingWhileDisconnected;
public TopicListWatcher(TopicListService topicListService, long id,
NamespaceName namespace, TopicsPattern topicsPattern, List<String> topics,
@@ -103,8 +112,8 @@ public class TopicListService {
* @param topicName topic name which contains partition suffix.
*/
@Override
- public synchronized void accept(String topicName, NotificationType notificationType) {
- if (closed || updatingTopics) {
+ public synchronized void onTopicEvent(String topicName, NotificationType notificationType) {
+ if (closed) {
return;
}
String partitionedTopicName = TopicName.get(topicName).getPartitionedTopicName();
@@ -117,7 +126,7 @@ public class TopicListService {
if (matchingTopics.remove(topicName)) {
deletedTopics = Collections.singletonList(topicName);
}
- } else if (matchingTopics.add(topicName)) {
+ } else if (notificationType == NotificationType.Created && matchingTopics.add(topicName)) {
newTopics = Collections.singletonList(topicName);
}
if (!newTopics.isEmpty() || !deletedTopics.isEmpty()) {
@@ -129,7 +138,7 @@ public class TopicListService {
// sends updates one-by-one so that ordering is retained
private synchronized void sendTopicListUpdate(String hash, List<String> deletedTopics, List<String> newTopics) {
- if (closed || updatingTopics) {
+ if (closed) {
return;
}
Runnable task = () -> topicListService.sendTopicListUpdate(id, hash, deletedTopics, newTopics,
@@ -140,12 +149,9 @@ public class TopicListService {
} else {
// if sendTopicListSuccess hasn't completed, add to a queue to be executed after it completes
if (!sendTopicListUpdateTasks.offer(task)) {
- log.warn("Update queue was full for watcher id {} matching {}. Performing full refresh.", id,
- topicsPattern.inputPattern());
- if (!updatingTopics) {
- updatingTopics = true;
- sendTopicListUpdateTasks.clear();
- matchingTopics.clear();
+ if (prepareUpdateTopics(null)) {
+ log.warn("Update queue was full for watcher id {} matching {}. Performing full refresh.", id,
+ topicsPattern.inputPattern());
executor.execute(() -> topicListService.updateTopicListWatcher(this));
}
}
@@ -171,12 +177,112 @@ public class TopicListService {
public synchronized void close() {
closed = true;
sendTopicListUpdateTasks.clear();
+ updateCallbacks.clear();
+ }
+
+ /**
+ * Returns true if the topic list update is prepared for execution. It is expected that the caller initiates
+ * the update. The callback is registered to be executed after the update, either existing or upcoming is
+ * completed.
+ * @param afterUpdateCompletionCallback callback to be executed after the update is completed.
+ * @return true if an existing update wasn't ongoing and a new update is prepared for execution.
+ */
+ synchronized boolean prepareUpdateTopics(Runnable afterUpdateCompletionCallback) {
+ if (!updatingTopics) {
+ updatingTopics = true;
+ sendingInProgress = true;
+ sendTopicListUpdateTasks.clear();
+ matchingTopics.clear();
+ if (afterUpdateCompletionCallback != null) {
+ updateCallbacks.add(afterUpdateCompletionCallback);
+ }
+ return true;
+ } else {
+ if (afterUpdateCompletionCallback != null) {
+ updateCallbacks.add(afterUpdateCompletionCallback);
+ }
+ return false;
+ }
}
synchronized void updateTopics(List<String> topics) {
+ if (closed) {
+ return;
+ }
matchingTopics.clear();
TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
updatingTopics = false;
+ if (disconnected) {
+ handleNewAndDeletedTopicsWhileDisconnected();
+ matchingTopicsBeforeDisconnected = null;
+ disconnected = false;
+ }
+ for (Runnable callback : updateCallbacks) {
+ try {
+ callback.run();
+ } catch (Exception e) {
+ log.warn("Error executing topic list update callback: {}", callback, e);
+ }
+ }
+ updateCallbacks.clear();
+ sendingCompleted();
+ }
+
+ private synchronized void handleNewAndDeletedTopicsWhileDisconnected() {
+ if (matchingTopicsBeforeDisconnected == null) {
+ return;
+ }
+ List<String> newTopics = new ArrayList<>();
+ List<String> deletedTopics = new ArrayList<>();
+ Set<String> remainingTopics = new HashSet<>(matchingTopics);
+ for (String topic : matchingTopicsBeforeDisconnected) {
+ if (!remainingTopics.remove(topic)) {
+ deletedTopics.add(topic);
+ }
+ }
+ newTopics.addAll(remainingTopics);
+ if (!newTopics.isEmpty() || !deletedTopics.isEmpty()) {
+ String hash = TopicList.calculateHash(matchingTopics);
+ sendTopicListUpdate(hash, deletedTopics, newTopics);
+ }
+ }
+
+ @Override
+ public NamespaceName getNamespaceName() {
+ return namespace;
+ }
+
+ @Override
+ public synchronized void onSessionEvent(SessionEvent event) {
+ switch (event) {
+ case SessionReestablished:
+ case Reconnected:
+ executor.execute(() -> {
+ synchronized (this) {
+ // ensure that only one update is triggered when connection is being lost and reconnected
+ // before the updating is complete. The updatingWhileDisconnected flag is reseted after
+ // the update completes.
+ if (!updatingWhileDisconnected) {
+ updatingWhileDisconnected = true;
+ CompletableFuture<Void> future = topicListService.updateTopicListWatcher(this);
+ future.whenComplete((__, ___) -> {
+ synchronized (this) {
+ updatingWhileDisconnected = false;
+ }
+ });
+ }
+ }
+ });
+ break;
+ case SessionLost:
+ case ConnectionLost:
+ if (!disconnected) {
+ disconnected = true;
+ matchingTopicsBeforeDisconnected = new ArrayList<>(matchingTopics);
+ prepareUpdateTopics(null);
+ }
+ break;
+ }
}
}
@@ -266,9 +372,27 @@ public class TopicListService {
CompletableFuture<TopicListWatcher> existingWatcherFuture = watchers.putIfAbsent(watcherId, watcherFuture);
if (existingWatcherFuture != null) {
- log.info("[{}] Watcher with the same watcherId={} is already created.", connection, watcherId);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Watcher with the same watcherId={} is already created. Refreshing.", connection,
+ watcherId);
+ }
// use the existing watcher if it's already created
- watcherFuture = existingWatcherFuture;
+ watcherFuture = existingWatcherFuture.thenCompose(watcher -> {
+ CompletableFuture<TopicListWatcher> future = new CompletableFuture<>();
+ Runnable callback = () -> future.complete(watcher);
+ // trigger a new update unless an update is already ongoing. Register the callback to complete
+ // when the update completes.
+ if (watcher.prepareUpdateTopics(callback)) {
+ updateTopicListWatcher(watcher)
+ // run the callback also in failure cases
+ // prepareUpdateTopics handles it for success cases
+ .exceptionally(ex -> {
+ callback.run();
+ return null;
+ });
+ }
+ return future;
+ });
} else {
initializeTopicsListWatcher(watcherFuture, namespaceName, watcherId, topicsPattern);
}
@@ -290,8 +414,8 @@ public class TopicListService {
lookupSemaphore.release();
})
.exceptionally(ex -> {
- log.warn("[{}] Error WatchTopicList for namespace [//{}] by {}",
- connection.toString(), namespaceName, requestId);
+ log.warn("[{}] Error WatchTopicList for namespace [//{}] by {}: {}",
+ connection.toString(), namespaceName, requestId, ex.getMessage());
connection.getCommandSender().sendErrorResponse(requestId,
BrokerServiceException.getClientErrorCode(
new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
@@ -327,130 +451,139 @@ public class TopicListService {
*/
public void initializeTopicsListWatcher(CompletableFuture<TopicListWatcher> watcherFuture,
NamespaceName namespace, long watcherId, TopicsPattern topicsPattern) {
- BooleanSupplier isPermitRequestCancelled = () -> !connection.isActive() || !watchers.containsKey(watcherId);
- if (isPermitRequestCancelled.getAsBoolean()) {
- return;
- }
- TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar.getBrokerService().getTopicListSizeResultCache()
- .getTopicListSize(namespace.toString(), CommandGetTopicsOfNamespace.Mode.PERSISTENT);
- AsyncDualMemoryLimiter maxTopicListInFlightLimiter = pulsar.getBrokerService().getMaxTopicListInFlightLimiter();
-
- listSizeHolder.getSizeAsync().thenCompose(initialSize -> {
- // use heap size limiter to avoid broker getting overwhelmed by a lot of concurrent topic list requests
- return maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
- AtomicReference<TopicListWatcher> watcherRef = new AtomicReference<>();
- return namespaceService.getListOfPersistentTopics(namespace).thenCompose(topics -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topics);
- listSizeHolder.updateSize(actualSize);
- // register watcher immediately so that we don't lose events
- TopicListWatcher watcher =
- new TopicListWatcher(this, watcherId, namespace, topicsPattern, topics,
- connection.ctx().executor(), topicListUpdateMaxQueueSize);
- watcherRef.set(watcher);
- topicResources.registerPersistentTopicListener(namespace, watcher);
- // use updated permits to slow down responses so that backpressure gets applied
- return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
- isPermitRequestCancelled, updatedPermits -> {
- // reset retry backoff
- retryBackoff.reset();
- // just return the watcher which was already created before
- return CompletableFuture.completedFuture(watcher);
- }, CompletableFuture::failedFuture);
- }).whenComplete((watcher, exception) -> {
- if (exception != null) {
- TopicListWatcher w = watcherRef.get();
- if (w != null) {
- w.close();
- topicResources.deregisterPersistentTopicListener(w);
- }
- // triggers a retry
- throw FutureUtil.wrapToCompletionException(exception);
- } else {
- if (!watcherFuture.complete(watcher)) {
- log.warn("[{}] Watcher future was already completed. Deregistering "
- + "watcherId={}.", connection, watcherId);
- watcher.close();
- topicResources.deregisterPersistentTopicListener(watcher);
- watchers.remove(watcherId, watcherFuture);
- }
- }
- });
- }, CompletableFuture::failedFuture);
- }).exceptionally(t -> {
- Throwable unwrappedException = FutureUtil.unwrapCompletionException(t);
- if (!isPermitRequestCancelled.getAsBoolean() && (
- unwrappedException instanceof AsyncSemaphore.PermitAcquireTimeoutException
- || unwrappedException instanceof AsyncSemaphore.PermitAcquireQueueFullException)) {
- // retry with backoff if permit acquisition fails due to timeout or queue full
- long retryAfterMillis = this.retryBackoff.next();
- log.info("[{}] {} when initializing topic list watcher watcherId={} for namespace {}. Retrying in {} "
- + "ms.", connection, unwrappedException.getMessage(), watcherId, namespace,
- retryAfterMillis);
- connection.ctx().executor()
- .schedule(() -> initializeTopicsListWatcher(watcherFuture, namespace, watcherId, topicsPattern),
- retryAfterMillis, TimeUnit.MILLISECONDS);
+ AtomicReference<TopicListWatcher> watcherRef = new AtomicReference<>();
+ Consumer<List<String>> afterListing = topics -> {
+ // register watcher immediately so that we don't lose events
+ TopicListWatcher watcher =
+ new TopicListWatcher(this, watcherId, namespace, topicsPattern, topics,
+ connection.ctx().executor(), topicListUpdateMaxQueueSize);
+ watcherRef.set(watcher);
+ topicResources.registerPersistentTopicListener(watcher);
+ };
+ getTopics(namespace, watcherId, afterListing).whenComplete((topics, exception) -> {
+ TopicListWatcher w = watcherRef.get();
+ if (exception != null) {
+ if (w != null) {
+ w.close();
+ topicResources.deregisterPersistentTopicListener(w);
+ }
+ Throwable unwrappedException = FutureUtil.unwrapCompletionException(exception);
+ if (connection.isActive() && (unwrappedException instanceof AsyncSemaphore.PermitAcquireTimeoutException
+ || unwrappedException instanceof AsyncSemaphore.PermitAcquireQueueFullException)) {
+ // retry with backoff if permit acquisition fails due to timeout or queue full
+ long retryAfterMillis = this.retryBackoff.next();
+ log.info("[{}] {} when initializing topic list watcher watcherId={} for namespace {}. "
+ + "Retrying in {} " + "ms.", connection, unwrappedException.getMessage(), watcherId,
+ namespace, retryAfterMillis);
+ connection.ctx().executor().schedule(
+ () -> initializeTopicsListWatcher(watcherFuture, namespace, watcherId, topicsPattern),
+ retryAfterMillis, TimeUnit.MILLISECONDS);
+ } else {
+ log.warn("[{}] Failed to initialize topic list watcher watcherId={} for namespace {}.", connection,
+ watcherId, namespace, unwrappedException);
+ watcherFuture.completeExceptionally(unwrappedException);
+ }
} else {
- log.warn("[{}] Failed to initialize topic list watcher watcherId={} for namespace {}.", connection,
- watcherId, namespace, unwrappedException);
- watcherFuture.completeExceptionally(unwrappedException);
+ if (!watcherFuture.complete(w)) {
+ log.warn("[{}] Watcher future was already completed. Deregistering " + "watcherId={}.", connection,
+ watcherId);
+ w.close();
+ topicResources.deregisterPersistentTopicListener(w);
+ watchers.remove(watcherId, watcherFuture);
+ }
}
- return null;
});
}
- void updateTopicListWatcher(TopicListWatcher watcher) {
- long watcherId = watcher.id;
+ private CompletableFuture<List<String>> getTopics(NamespaceName namespace, long watcherId) {
+ return getTopics(namespace, watcherId, null);
+ }
+
+ private CompletableFuture<List<String>> getTopics(NamespaceName namespace, long watcherId,
+ Consumer<List<String>> afterListing) {
BooleanSupplier isPermitRequestCancelled = () -> !connection.isActive() || !watchers.containsKey(watcherId);
if (isPermitRequestCancelled.getAsBoolean()) {
- return;
+ return CompletableFuture.failedFuture(
+ new AsyncSemaphore.PermitAcquireCancelledException("Permit acquisition was cancelled"));
}
- NamespaceName namespace = watcher.namespace;
+ return getTopics(namespace, afterListing, isPermitRequestCancelled);
+ }
+
+ private CompletableFuture<List<String>> getTopics(NamespaceName namespace,
+ Consumer<List<String>> afterListing,
+ BooleanSupplier isPermitRequestCancelled) {
TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar.getBrokerService().getTopicListSizeResultCache()
.getTopicListSize(namespace.toString(), CommandGetTopicsOfNamespace.Mode.PERSISTENT);
AsyncDualMemoryLimiter maxTopicListInFlightLimiter = pulsar.getBrokerService().getMaxTopicListInFlightLimiter();
- listSizeHolder.getSizeAsync().thenCompose(initialSize -> {
+ return listSizeHolder.getSizeAsync().thenCompose(initialSize -> {
// use heap size limiter to avoid broker getting overwhelmed by a lot of concurrent topic list requests
return maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
- return namespaceService.getListOfPersistentTopics(namespace).thenCompose(topics -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topics);
- listSizeHolder.updateSize(actualSize);
- // use updated permits to slow down responses so that backpressure gets applied
- return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
- isPermitRequestCancelled, updatedPermits -> {
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ return namespaceService.getListOfUserTopics(namespace,
+ CommandGetTopicsOfNamespace.Mode.PERSISTENT).thenComposeAsync(topics -> {
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topics);
+ listSizeHolder.updateSize(actualSize);
+ if (afterListing != null) {
+ afterListing.accept(topics);
+ }
+ if (initialSize != actualSize) {
+ // use updated permits to slow down responses so that backpressure gets applied
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits,
+ actualSize,
+ isPermitRequestCancelled, updatedPermits -> {
+ // reset retry backoff
+ retryBackoff.reset();
+ // just return the topics which were already retrieved before
+ return CompletableFuture.completedFuture(topics);
+ }, CompletableFuture::failedFuture);
+ } else {
// reset retry backoff
retryBackoff.reset();
- // just return topics here
return CompletableFuture.completedFuture(topics);
- }, CompletableFuture::failedFuture);
- }).whenComplete((topics, exception) -> {
- if (exception != null) {
- // triggers a retry
- throw FutureUtil.wrapToCompletionException(exception);
- } else {
- watcher.updateTopics(topics);
- }
- });
- }, CompletableFuture::failedFuture);
- }).exceptionally(t -> {
- Throwable unwrappedException = FutureUtil.unwrapCompletionException(t);
- if (!isPermitRequestCancelled.getAsBoolean() && (
- unwrappedException instanceof AsyncSemaphore.PermitAcquireTimeoutException
- || unwrappedException instanceof AsyncSemaphore.PermitAcquireQueueFullException)) {
- // retry with backoff if permit acquisition fails due to timeout or queue full
- long retryAfterMillis = this.retryBackoff.next();
- log.info("[{}] {} when updating topic list watcher watcherId={} for namespace {}. Retrying in {} "
- + "ms.", connection, unwrappedException.getMessage(), watcherId, namespace,
- retryAfterMillis);
- connection.ctx().executor()
- .schedule(() -> updateTopicListWatcher(watcher), retryAfterMillis, TimeUnit.MILLISECONDS);
+ }
+ }, connection.ctx().executor());
+ }, CompletableFuture::failedFuture)
+ .thenApplyAsync(Function.identity(), connection.ctx().executor());
+ });
+ }
+
+ CompletableFuture<Void> updateTopicListWatcher(TopicListWatcher watcher) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ try {
+ internalUpdateTopicListWatcher(watcher, future);
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ void internalUpdateTopicListWatcher(TopicListWatcher watcher, CompletableFuture<Void> future) {
+ NamespaceName namespace = watcher.namespace;
+ long watcherId = watcher.id;
+ getTopics(namespace, watcherId).whenComplete((topics, exception) -> {
+ if (exception != null) {
+ Throwable unwrappedException = FutureUtil.unwrapCompletionException(exception);
+ if (connection.isActive() && !watcher.closed
+ && (unwrappedException instanceof AsyncSemaphore.PermitAcquireTimeoutException
+ || unwrappedException instanceof AsyncSemaphore.PermitAcquireQueueFullException)) {
+ // retry with backoff if permit acquisition fails due to timeout or queue full
+ long retryAfterMillis = this.retryBackoff.next();
+ log.info("[{}] {} when updating topic list watcher watcherId={} for namespace {}. Retrying in {} "
+ + "ms.", connection, unwrappedException.getMessage(), watcherId, namespace,
+ retryAfterMillis);
+ connection.ctx().executor()
+ .schedule(() -> internalUpdateTopicListWatcher(watcher, future),
+ retryAfterMillis, TimeUnit.MILLISECONDS);
+ } else {
+ log.warn("[{}] Failed to update topic list watcher watcherId={} for namespace {}.", connection,
+ watcherId, namespace, unwrappedException);
+ future.completeExceptionally(unwrappedException);
+ }
} else {
- log.warn("[{}] Failed to update topic list watcher watcherId={} for namespace {}.", connection,
- watcherId, namespace, unwrappedException);
+ watcher.updateTopics(topics);
+ future.complete(null);
}
- return null;
});
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index a19bc017574..9044f4a910f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -74,6 +74,7 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
import javax.naming.AuthenticationException;
import lombok.AllArgsConstructor;
+import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
@@ -235,7 +236,7 @@ public class ServerCnxTest {
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfTopics(
NamespaceName.get("use", "ns-abc"), CommandGetTopicsOfNamespace.Mode.ALL);
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfUserTopics(
- NamespaceName.get("use", "ns-abc"), CommandGetTopicsOfNamespace.Mode.ALL);
+ eq(NamespaceName.get("use", "ns-abc")), any());
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfPersistentTopics(
NamespaceName.get("use", "ns-abc"));
doReturn(CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists())).when(namespaceService)
@@ -3253,6 +3254,8 @@ public class ServerCnxTest {
public void testWatchTopicList() throws Exception {
svcConfig.setEnableBrokerSideSubscriptionPatternEvaluation(true);
resetChannel();
+ @Cleanup
+ BackGroundExecutor backGroundExecutor = startBackgroundExecutorForEmbeddedChannel(channel);
setChannelConnected();
BaseCommand command = Commands.newWatchTopicList(1, 3, "use/ns-abc", "use/ns-abc/topic-.*", null);
ByteBuf serializedCommand = Commands.serializeWithSize(command);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java
index d85f244b9a3..5caa7712891 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java
@@ -21,8 +21,10 @@ package org.apache.pulsar.broker.service;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
@@ -43,6 +45,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -62,17 +65,20 @@ import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.resources.TopicResources;
import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
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.semaphore.AsyncDualMemoryLimiter;
import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiterImpl;
import org.apache.pulsar.common.semaphore.AsyncSemaphore;
import org.apache.pulsar.common.topics.TopicList;
import org.apache.pulsar.common.topics.TopicsPattern;
-import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
import org.awaitility.Awaitility;
import org.jspecify.annotations.NonNull;
import org.mockito.InOrder;
@@ -97,6 +103,7 @@ public class TopicListServiceTest {
private AsyncDualMemoryLimiterImpl memoryLimiter;
private ScheduledExecutorService scheduledExecutorService;
private PulsarService pulsar;
+ private NamespaceService namespaceService;
@BeforeMethod(alwaysRun = true)
public void setup() throws Exception {
@@ -105,7 +112,7 @@ public class TopicListServiceTest {
topicListFuture = new CompletableFuture<>();
AtomicReference<Consumer<Notification>> listenerRef = new AtomicReference<>();
- MetadataStore metadataStore = mock(MetadataStore.class);
+ MetadataStoreExtended metadataStore = mock(MetadataStoreExtended.class);
doAnswer(invocationOnMock -> {
listenerRef.set(invocationOnMock.getArgument(0));
return null;
@@ -114,10 +121,10 @@ public class TopicListServiceTest {
notificationConsumer = listenerRef.get();
pulsar = mock(PulsarService.class);
- NamespaceService namespaceService = mock(NamespaceService.class);
+ namespaceService = mock(NamespaceService.class);
when(pulsar.getNamespaceService()).thenReturn(namespaceService);
doAnswer(invocationOnMock -> topicListFuture)
- .when(namespaceService).getListOfPersistentTopics(any());
+ .when(namespaceService).getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
when(pulsar.getPulsarResources()).thenReturn(mock(PulsarResources.class));
when(pulsar.getPulsarResources().getTopicResources()).thenReturn(topicResources);
@@ -200,8 +207,7 @@ public class TopicListServiceTest {
String hash = TopicList.calculateHash(topics);
topicListFuture.complete(topics);
Awaitility.await().untilAsserted(() -> Assert.assertEquals(1, lookupSemaphore.availablePermits()));
- verify(topicResources).registerPersistentTopicListener(
- eq(NamespaceName.get("tenant/ns")), any(TopicListService.TopicListWatcher.class));
+ verify(topicResources).registerPersistentTopicListener(any(TopicListService.TopicListWatcher.class));
Collection<String> expectedTopics = new ArrayList<>(topics);
verify(connection.getCommandSender()).sendWatchTopicListSuccess(eq(7L), eq(13L), eq(hash), eq(expectedTopics),
any());
@@ -229,8 +235,7 @@ public class TopicListServiceTest {
// release the permits
memoryLimiter.release(permit);
Awaitility.await().untilAsserted(() -> Assert.assertEquals(1, lookupSemaphore.availablePermits()));
- verify(topicResources).registerPersistentTopicListener(
- eq(NamespaceName.get("tenant/ns")), any(TopicListService.TopicListWatcher.class));
+ verify(topicResources).registerPersistentTopicListener(any(TopicListService.TopicListWatcher.class));
Collection<String> expectedTopics = new ArrayList<>(topics);
verify(connection.getCommandSender()).sendWatchTopicListSuccess(eq(7L), eq(13L), eq(hash), eq(expectedTopics),
any());
@@ -382,19 +387,15 @@ public class TopicListServiceTest {
topicListFuture.complete(topics);
assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
- CompletableFuture<Void> completePending = new CompletableFuture<>();
- doReturn(completePending).when(pulsarCommandSender)
- .sendWatchTopicListUpdate(anyLong(), any(), any(), anyString(), any());
topicListFuture = new CompletableFuture<>();
-
+ List<String> updatedTopics = IntStream.range(1, 101).mapToObj(i -> "persistent://tenant/ns/topic" + i).toList();
// when the queue overflows
- for (int i = 10; i <= 10 + topicListUpdateMaxQueueSize + 1; i++) {
- notificationConsumer.accept(
- new Notification(NotificationType.Created, "/managed-ledgers/tenant/ns/persistent/topic" + i));
+ for (int i = 1; i < updatedTopics.size(); i++) {
+ TopicName topicName = TopicName.get(updatedTopics.get(i));
+ notificationConsumer.accept(new Notification(NotificationType.Created,
+ "/managed-ledgers/" + topicName.getPersistenceNamingEncoding()));
}
-
- // a new listing should be performed. Return 100 topics in the response, simulating that events have been lost
- List<String> updatedTopics = IntStream.range(1, 101).mapToObj(i -> "persistent://tenant/ns/topic" + i).toList();
+ // a new listing should be performed
topicListFuture.complete(updatedTopics);
// validate that the watcher's matching topics have been updated
Awaitility.await().untilAsserted(() -> {
@@ -405,4 +406,330 @@ public class TopicListServiceTest {
});
}
+ @Test
+ public void testCommandWatchSuccessNoTopicsInResponseWhenHashMatches() {
+ List<String> topics = Collections.singletonList("persistent://tenant/ns/topic1");
+ String hash = TopicList.calculateHash(topics);
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, hash,
+ lookupSemaphore);
+ doReturn(CompletableFuture.completedFuture(null)).when(pulsarCommandSender)
+ .sendWatchTopicListSuccess(anyLong(), anyLong(), anyString(), any(), any());
+ topicListFuture.complete(topics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+ Collection<String> expectedTopics = List.of();
+ verify(connection.getCommandSender(), timeout(2000L).times(1))
+ .sendWatchTopicListSuccess(eq(7L), eq(13L), eq(hash), eq(expectedTopics), any());
+ }
+
+ @Test
+ public void testCommandWatchSuccessTopicsInResponseWhenHashDoesntMatch() {
+ List<String> topics = Collections.singletonList("persistent://tenant/ns/topic1");
+ String hash = TopicList.calculateHash(topics);
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, "INVALID_HASH",
+ lookupSemaphore);
+ doReturn(CompletableFuture.completedFuture(null)).when(pulsarCommandSender)
+ .sendWatchTopicListSuccess(anyLong(), anyLong(), anyString(), any(), any());
+ topicListFuture.complete(topics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+ Collection<String> expectedTopics = topics;
+ verify(connection.getCommandSender(), timeout(2000L).times(1))
+ .sendWatchTopicListSuccess(eq(7L), eq(13L), eq(hash), eq(expectedTopics), any());
+ }
+
+ @Test
+ public void testSessionDisconnectAndReconnectSendsNewAndDeletedTopics() {
+ // Initial topics: topic1, topic2, topic3
+ List<String> initialTopics = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2",
+ "persistent://tenant/ns/topic3");
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, null,
+ lookupSemaphore);
+ topicListFuture.complete(initialTopics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+
+ TopicListService.TopicListWatcher watcher = topicListService.getWatcherFuture(13).join();
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(initialTopics);
+
+ // Simulate session disconnect
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+
+ // Prepare topics after reconnect: topic2 remains, topic1 and topic3 deleted, topic4 and topic5 added
+ List<String> topicsAfterReconnect = List.of(
+ "persistent://tenant/ns/topic2",
+ "persistent://tenant/ns/topic4",
+ "persistent://tenant/ns/topic5");
+
+ topicListFuture = new CompletableFuture<>();
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ doAnswer(invocationOnMock -> topicListFuture)
+ .when(namespaceService).getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
+
+ // Simulate session reconnect - this should trigger a topic list refresh
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+
+ // Complete the topic list future with the new topics
+ topicListFuture.complete(topicsAfterReconnect);
+
+ // Expected: deleted topics are topic1 and topic3, new topics are topic4 and topic5
+ List<String> expectedDeleted = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic3");
+ List<String> expectedNew = List.of(
+ "persistent://tenant/ns/topic4",
+ "persistent://tenant/ns/topic5");
+ String expectedHash = TopicList.calculateHash(topicsAfterReconnect);
+
+ // Verify that sendWatchTopicListUpdate is called with the correct new and deleted topics
+ verify(connection.getCommandSender(), timeout(2000L))
+ .sendWatchTopicListUpdate(eq(13L),
+ argThat(newTopics -> new HashSet<>(newTopics).equals(new HashSet<>(expectedNew))),
+ argThat(deletedTopics -> new HashSet<>(deletedTopics).equals(new HashSet<>(expectedDeleted))),
+ eq(expectedHash), any());
+
+ // Verify the watcher's matching topics are updated
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(topicsAfterReconnect);
+ }
+
+ @Test
+ public void testSessionLostAndReestablishedSendsNewAndDeletedTopics() {
+ // Initial topics: topic1, topic2
+ List<String> initialTopics = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2");
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, null,
+ lookupSemaphore);
+ topicListFuture.complete(initialTopics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+
+ TopicListService.TopicListWatcher watcher = topicListService.getWatcherFuture(13).join();
+
+ // Simulate session lost (more severe than connection lost)
+ watcher.onSessionEvent(SessionEvent.SessionLost);
+
+ // After reconnect: all topics deleted, completely new topics added
+ List<String> topicsAfterReconnect = List.of(
+ "persistent://tenant/ns/topic7",
+ "persistent://tenant/ns/topic8",
+ "persistent://tenant/ns/topic9");
+
+ topicListFuture = new CompletableFuture<>();
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ doAnswer(invocationOnMock -> topicListFuture)
+ .when(namespaceService).getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
+
+ // Simulate session reestablished
+ watcher.onSessionEvent(SessionEvent.SessionReestablished);
+
+ topicListFuture.complete(topicsAfterReconnect);
+
+ List<String> expectedDeleted = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2");
+ List<String> expectedNew = List.of(
+ "persistent://tenant/ns/topic7",
+ "persistent://tenant/ns/topic8",
+ "persistent://tenant/ns/topic9");
+ String expectedHash = TopicList.calculateHash(topicsAfterReconnect);
+
+ verify(connection.getCommandSender(), timeout(2000L))
+ .sendWatchTopicListUpdate(eq(13L),
+ argThat(newTopics -> new HashSet<>(newTopics).equals(new HashSet<>(expectedNew))),
+ argThat(deletedTopics -> new HashSet<>(deletedTopics).equals(new HashSet<>(expectedDeleted))),
+ eq(expectedHash), any());
+
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(topicsAfterReconnect);
+ }
+
+ @Test
+ public void testSessionReconnectWithNoChangesDoesNotSendUpdate() {
+ // Initial topics
+ List<String> initialTopics = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2");
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, null,
+ lookupSemaphore);
+ topicListFuture.complete(initialTopics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+
+ TopicListService.TopicListWatcher watcher = topicListService.getWatcherFuture(13).join();
+
+ // Simulate session disconnect
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+
+ // Topics remain the same after reconnect
+ topicListFuture = new CompletableFuture<>();
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ doAnswer(invocationOnMock -> topicListFuture)
+ .when(namespaceService).getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
+
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+
+ // Complete with the same topics
+ topicListFuture.complete(initialTopics);
+
+ // Wait for processing
+ Awaitility.await().untilAsserted(() ->
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(initialTopics));
+
+ // Verify that sendWatchTopicListUpdate was NOT called (no changes to report)
+ verify(connection.getCommandSender(), timeout(500L).times(0))
+ .sendWatchTopicListUpdate(eq(13L), any(), any(), any(), any());
+ }
+
+ @Test
+ public void testConnectionOrSessionFlappingDoesNotTriggerOverlappingUpdate() {
+ List<String> initialTopics = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2");
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, null,
+ lookupSemaphore);
+ topicListFuture.complete(initialTopics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+
+ TopicListService.TopicListWatcher watcher = topicListService.getWatcherFuture(13).join();
+ clearInvocations(namespaceService);
+
+ topicListFuture = new CompletableFuture<>();
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+
+ Awaitility.await().untilAsserted(() -> verify(namespaceService, timeout(500L).times(1))
+ .getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT)));
+
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+ watcher.onSessionEvent(SessionEvent.SessionLost);
+ watcher.onSessionEvent(SessionEvent.SessionReestablished);
+
+ verify(namespaceService, timeout(500L).times(1))
+ .getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
+
+ topicListFuture.complete(initialTopics);
+ Awaitility.await().untilAsserted(() ->
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(initialTopics));
+ }
+
+ @Test
+ public void testWatchTopicListReconcileDoesntTriggerOverlappingUpdate() {
+ List<String> initialTopics = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2");
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, null,
+ lookupSemaphore);
+ topicListFuture.complete(initialTopics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+
+ TopicListService.TopicListWatcher watcher = topicListService.getWatcherFuture(13).join();
+
+ clearInvocations(namespaceService);
+ topicListFuture = new CompletableFuture<>();
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+ Awaitility.await().untilAsserted(() -> verify(namespaceService, timeout(500L).times(1))
+ .getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT)));
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, TopicList.calculateHash(initialTopics),
+ lookupSemaphore);
+
+ topicListFuture.complete(initialTopics);
+
+ verify(namespaceService, timeout(500L).times(1))
+ .getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(initialTopics));
+ }
+
+ @Test
+ public void testWatchTopicListReconcileNoOverlappingUpdate() {
+ List<String> initialTopics = List.of(
+ "persistent://tenant/ns/topic1",
+ "persistent://tenant/ns/topic2");
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, null,
+ lookupSemaphore);
+ topicListFuture.complete(initialTopics);
+ assertThat(topicListService.getWatcherFuture(13)).succeedsWithin(Duration.ofSeconds(2));
+
+ TopicListService.TopicListWatcher watcher = topicListService.getWatcherFuture(13).join();
+
+ clearInvocations(namespaceService);
+ topicListFuture = new CompletableFuture<>();
+ watcher.onSessionEvent(SessionEvent.ConnectionLost);
+ watcher.onSessionEvent(SessionEvent.Reconnected);
+ Awaitility.await().untilAsserted(() -> verify(namespaceService, timeout(500L).times(1))
+ .getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT)));
+
+ topicListFuture.complete(initialTopics);
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(watcher.isUpdatingTopics()).isFalse();
+ });
+
+ topicListService.handleWatchTopicList(
+ NamespaceName.get("tenant/ns"),
+ 13,
+ 7,
+ "persistent://tenant/ns/topic\\d",
+ topicsPatternImplementation, TopicList.calculateHash(initialTopics),
+ lookupSemaphore);
+
+ verify(namespaceService, timeout(1000L).times(2))
+ .getListOfUserTopics(any(), eq(CommandGetTopicsOfNamespace.Mode.PERSISTENT));
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(watcher.getMatchingTopics()).containsExactlyInAnyOrderElementsOf(initialTopics));
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java
index 46262e84d38..1a75dd70bae 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java
@@ -75,7 +75,7 @@ public class TopicListWatcherTest {
@Test
public void testAcceptSendsNotificationAndRemembersTopic() {
String newTopic = "persistent://tenant/ns/topic3";
- watcher.accept(newTopic, NotificationType.Created);
+ watcher.onTopicEvent(newTopic, NotificationType.Created);
List<String> allMatchingTopics = Arrays.asList(
"persistent://tenant/ns/topic1", "persistent://tenant/ns/topic2", newTopic);
@@ -90,7 +90,7 @@ public class TopicListWatcherTest {
@Test
public void testAcceptSendsNotificationAndForgetsTopic() {
String deletedTopic = "persistent://tenant/ns/topic1";
- watcher.accept(deletedTopic, NotificationType.Deleted);
+ watcher.onTopicEvent(deletedTopic, NotificationType.Deleted);
List<String> allMatchingTopics = Collections.singletonList("persistent://tenant/ns/topic2");
String hash = TopicList.calculateHash(allMatchingTopics);
@@ -103,7 +103,7 @@ public class TopicListWatcherTest {
@Test
public void testAcceptIgnoresNonMatching() {
- watcher.accept("persistent://tenant/ns/mytopic", NotificationType.Created);
+ watcher.onTopicEvent("persistent://tenant/ns/mytopic", NotificationType.Created);
verifyNoInteractions(topicListService);
Assert.assertEquals(
Arrays.asList("persistent://tenant/ns/topic1", "persistent://tenant/ns/topic2"),
@@ -114,7 +114,7 @@ public class TopicListWatcherTest {
public void testUpdateQueueOverFlowPerformsFullUpdate() {
for (int i = 10; i <= 20; i++) {
String newTopic = "persistent://tenant/ns/topic" + i;
- watcher.accept(newTopic, NotificationType.Created);
+ watcher.onTopicEvent(newTopic, NotificationType.Created);
}
verify(topicListService).sendTopicListUpdate(anyLong(), anyString(), any(), any(), any());
verify(topicListService).updateTopicListWatcher(any());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java
index 46e41be0125..7fba9c9fc28 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java
@@ -149,7 +149,7 @@ class NonStartableTestPulsarService extends AbstractTestPulsarService {
private final TopicResources topicResources;
private final NamespaceResources namespaceResources;
- public TestPulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore,
+ public TestPulsarResources(MetadataStoreExtended localMetadataStore, MetadataStore configurationMetadataStore,
TopicResources topicResources, NamespaceResources namespaceResources) {
super(localMetadataStore, configurationMetadataStore);
this.topicResources = topicResources;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternConsumerTopicWatcherBackPressureMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternConsumerTopicWatcherBackPressureMultipleConsumersTest.java
index 9255abe42b9..f78fcec69cf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternConsumerTopicWatcherBackPressureMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternConsumerTopicWatcherBackPressureMultipleConsumersTest.java
@@ -23,7 +23,6 @@ import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -32,12 +31,12 @@ import java.util.stream.IntStream;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.PatternMultiTopicsConsumerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.TopicListWatcher;
import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiter;
import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiterImpl;
import org.apache.pulsar.common.util.FutureUtil;
@@ -138,15 +137,11 @@ public class PatternConsumerTopicWatcherBackPressureMultipleConsumersTest extend
List<Consumer<String>> consumers = consumerFutures.stream().map(CompletableFuture::join).toList();
- List<? extends CompletableFuture<?>> watcherFutures = consumers.stream().map(consumer -> {
- try {
- CompletableFuture<?> watcherFuture = consumer instanceof PatternMultiTopicsConsumerImpl
- ? (CompletableFuture<?>) FieldUtils.readField(consumer, "watcherFuture", true) : null;
- return watcherFuture;
- } catch (IllegalAccessException e) {
- throw new RuntimeException(e);
- }
- }).filter(Objects::nonNull).toList();
+ List<CompletableFuture<TopicListWatcher>> watcherFutures =
+ consumers.stream().filter(PatternMultiTopicsConsumerImpl.class::isInstance)
+ .map(PatternMultiTopicsConsumerImpl.class::cast)
+ .map(c -> (CompletableFuture<TopicListWatcher>) c.getWatcherFuture())
+ .toList();
// wait for all watcher futures to complete
FutureUtil.waitForAll(watcherFutures).get(60, TimeUnit.SECONDS);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java
index b8855656ad0..63e4d513a69 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConSupports.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.assertj.core.api.Assertions.assertThat;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -191,4 +192,12 @@ public abstract class AutoCloseUselessClientConSupports extends MultiBrokerBaseT
Assert.assertEquals(new String(messageTx.getData(), StandardCharsets.UTF_8), messageContentTx);
consumer.acknowledge(messageTx);
}
+
+ protected void waitForTopicListWatcherStarted(Consumer<?> consumer) {
+ Awaitility.await().untilAsserted(() -> {
+ CompletableFuture<TopicListWatcher> completableFuture =
+ ((PatternMultiTopicsConsumerImpl) consumer).getWatcherFuture();
+ assertThat(completableFuture).describedAs("Topic list watcher future should be done").isDone();
+ });
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java
index 981612d3930..be4ca3c4a50 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConTopicsPatternConsumerTest.java
@@ -72,6 +72,9 @@ public class AutoCloseUselessClientConTopicsPatternConsumerTest extends AutoClos
.subscriptionName("my-subscription-y")
.subscribe();
+ waitForTopicListWatcherStarted(consumer);
+ waitForTopicListWatcherStarted(consumer2);
+
// check that there are more than 3 connections
// at least 3 connections are required:
// 1 for "producer", 1 for "consumer", and 1 for the topic watcher of "consumer"
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 c1b96a058f4..ab4ed0dd95e 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
@@ -18,9 +18,9 @@
*/
package org.apache.pulsar.client.impl;
-import static org.mockito.Mockito.doReturn;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -35,7 +35,6 @@ 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.InjectedClientCnxClientBuilder;
import org.apache.pulsar.client.api.Message;
@@ -55,7 +54,6 @@ 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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
@@ -200,11 +198,11 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer4 = pulsarClient.newProducer().topic(topicName4)
.enableBatching(false)
@@ -221,10 +219,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
assertTrue(consumer.getTopic().startsWith(PatternMultiTopicsConsumerImpl.DUMMY_TOPIC_NAME_PREFIX));
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
// 4. verify consumer get methods, to get right number of partitions and topics.
assertSame(pattern.pattern(), ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern().inputPattern());
@@ -305,17 +300,18 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
assertTrue(consumer.getTopic().startsWith(PatternMultiTopicsConsumerImpl.DUMMY_TOPIC_NAME_PREFIX));
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
// 4. verify consumer get methods, to get right number of partitions and topics.
assertSame(pattern.pattern(), ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern().inputPattern());
List<String> topics = ((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitions();
List<ConsumerImpl<byte[]>> consumers = ((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getConsumers();
- assertEquals(topics.size(), 6);
+ List<String> expectedTopics =
+ List.of(topicName1 + "-partition-0",
+ topicName2 + "-partition-0", topicName2 + "-partition-1",
+ topicName3 + "-partition-0", topicName3 + "-partition-1", topicName3 + "-partition-2");
+ assertThat(topics).containsExactlyInAnyOrderElementsOf(expectedTopics);
assertEquals(consumers.size(), 6);
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitionedTopics().size(), 3);
@@ -368,11 +364,11 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer4 = pulsarClient.newProducer().topic(topicName4)
.enableBatching(false)
@@ -388,10 +384,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.subscribe();
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
// 4. verify consumer get methods, to get right number of partitions and topics.
assertSame(pattern.pattern(), ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern().inputPattern());
@@ -466,11 +459,11 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer4 = pulsarClient.newProducer().topic(topicName4)
.enableBatching(false)
@@ -486,10 +479,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.subscribe();
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
// 4. verify consumer get methods, to get right number of partitions and topics.
assertSame(pattern.pattern(), ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern().inputPattern());
@@ -563,10 +553,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.receiverQueueSize(4)
.subscribe();
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
// 3. verify consumer get methods, to get 5 number of partitions and topics.
assertSame(pattern.pattern(), ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern().inputPattern());
@@ -584,11 +571,11 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
// 5. verify consumer get methods, to get number of partitions and topics, value 6=1+2+3.
@@ -694,10 +681,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.subscribe();
// 0. Need make sure topic watcher started
- waitForTopicListWatcherStarted(consumer);
-
- // if broker enable watch topic, then recheckPatternTimeout will be null.
- assertNull(((PatternMultiTopicsConsumerImpl<?>) consumer).getRecheckPatternTimeout());
+ waitTopicListWatcherCreation(consumer);
// 1. create partition topic
String topicName = "persistent://my-property/my-ns/test-pattern" + key;
@@ -717,6 +701,59 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
pulsarClient.close();
}
+ @Test(timeOut = testTimeout)
+ public void testPeriodicReconciliationWithActiveWatcher() throws Exception {
+ String key = "PeriodicReconciliationWithActiveWatcher";
+ String subscriptionName = "my-ex-subscription-" + key;
+ String topicName = "persistent://my-property/my-ns/pattern-topic-1-" + key;
+ Pattern pattern = Pattern.compile("persistent://my-property/my-ns/pattern-topic.*");
+
+ // Create a topic so consumer has something to subscribe to
+ admin.topics().createPartitionedTopic(topicName, 2);
+
+ // Create consumer with short auto-discovery period (1 second)
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topicsPattern(pattern)
+ .patternAutoDiscoveryPeriod(1, TimeUnit.SECONDS)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscribe();
+
+ try {
+ // Wait for topic list watcher to be connected
+ waitTopicListWatcherCreation(consumer);
+
+ PatternMultiTopicsConsumerImpl<?> patternConsumer = (PatternMultiTopicsConsumerImpl<?>) consumer;
+
+ // Verify that recheckPatternTimeout is NOT null even with active watcher
+ // This confirms the periodic timer is always scheduled
+ assertNotNull(patternConsumer.getRecheckPatternTimeout(),
+ "recheckPatternTimeout should not be null even when TopicListWatcher is connected");
+
+ // Record the initial recheck epoch
+ int initialEpoch = patternConsumer.getRecheckPatternEpoch();
+
+ // Wait for at least 2 reconciliation cycles (with 1 second period)
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(10))
+ .pollInterval(Duration.ofMillis(500))
+ .untilAsserted(() -> {
+ int currentEpoch = patternConsumer.getRecheckPatternEpoch();
+ assertTrue(currentEpoch > initialEpoch + 1,
+ "recheckPatternEpoch should increase over time, indicating periodic reconciliation. "
+ + "Initial: " + initialEpoch + ", Current: " + currentEpoch);
+ });
+
+ // Verify timeout is still scheduled after reconciliation cycles
+ assertNotNull(patternConsumer.getRecheckPatternTimeout(),
+ "recheckPatternTimeout should remain scheduled after reconciliation cycles");
+
+ } finally {
+ consumer.close();
+ admin.topics().deletePartitionedTopic(topicName);
+ }
+ }
+
@DataProvider(name = "topicDomain")
public Object[][] topicDomain(){
return new Object[][]{
@@ -788,9 +825,10 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
};
}
- private void waitForTopicListWatcherStarted(Consumer consumer) {
+ private void waitTopicListWatcherCreation(Consumer<?> consumer) {
Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
+ CompletableFuture<TopicListWatcher> completableFuture =
+ ((PatternMultiTopicsConsumerImpl) consumer).getWatcherFuture();
log.info("isDone: {}, isCompletedExceptionally: {}", completableFuture.isDone(),
completableFuture.isCompletedExceptionally());
assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
@@ -814,7 +852,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.receiverQueueSize(4)
.subscribe();
if (createTopicAfterWatcherStarted) {
- waitForTopicListWatcherStarted(consumer);
+ waitTopicListWatcherCreation(consumer);
}
// 1. create topic.
@@ -867,7 +905,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.topicsPattern(pattern)
// Disable brokerSideSubscriptionPatternEvaluation will leading disable topic list watcher.
// So set patternAutoDiscoveryPeriod to a little value.
- .patternAutoDiscoveryPeriod(1)
+ .patternAutoDiscoveryPeriod(5, TimeUnit.SECONDS)
.subscriptionName(subscriptionName)
.subscriptionType(SubscriptionType.Shared)
.ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
@@ -882,9 +920,9 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
}
// 2. verify consumer can subscribe the topic.
- // Since the minimum value of `patternAutoDiscoveryPeriod` is 60s, we set the test timeout to a triple value.
+ // Since the minimum value of `patternAutoDiscoveryPeriod` is 5s, we set the test timeout to a triple value.
assertSame(pattern.pattern(), ((PatternMultiTopicsConsumerImpl<?>) consumer).getPattern().inputPattern());
- Awaitility.await().atMost(Duration.ofMinutes(3)).untilAsserted(() -> {
+ Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getPartitions().size(), 1);
assertEquals(((PatternMultiTopicsConsumerImpl<?>) consumer).getConsumers().size(), 1);
if (partitioned) {
@@ -955,11 +993,11 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Consumer<byte[]> consumer = pulsarClient.newConsumer()
@@ -972,10 +1010,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.subscribe();
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
assertTrue(consumer instanceof PatternMultiTopicsConsumerImpl);
@@ -1008,7 +1043,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
admin.topics().createPartitionedTopic(topicName4, 4);
Producer<byte[]> producer4 = pulsarClient.newProducer().topic(topicName4)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
// 7. call recheckTopics to subscribe each added topics above, verify topics number: 10=1+2+3+4
@@ -1070,27 +1105,24 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.enableBatching(false)
- .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
Consumer<byte[]> consumer = pulsarClient.newConsumer()
- .topicsPattern(pattern)
- .patternAutoDiscoveryPeriod(10, TimeUnit.SECONDS)
- .subscriptionName(subscriptionName)
- .subscriptionType(SubscriptionType.Shared)
- .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
- .receiverQueueSize(4)
- .subscribe();
+ .topicsPattern(pattern)
+ .patternAutoDiscoveryPeriod(3, TimeUnit.SECONDS)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
+ .receiverQueueSize(4)
+ .subscribe();
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
assertTrue(consumer instanceof PatternMultiTopicsConsumerImpl);
@@ -1118,23 +1150,27 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
} while (message != null);
assertEquals(messageSet, totalMessages);
- // 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.
- 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"));
+ // 6. remove producer 1,3 and delete topics; verify only consumer 2 left
+ producer3.close();
+ producer1.close();
// 7. call recheckTopics to unsubscribe topic 1,3, verify topics number: 2=6-1-3
log.debug("recheck topics change");
- 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);
- assertEquals(((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getPartitionedTopics().size(), 1);
+ Awaitility.await().untilAsserted(() -> {
+ try {
+ admin.topics().deletePartitionedTopic(topicName3, true);
+ } catch (Exception e) {
+ // ignore
+ }
+ try {
+ admin.topics().delete(topicName1, true);
+ } catch (Exception e) {
+ // ignore
+ }
+ assertEquals(((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getPartitions().size(), 2);
+ assertEquals(((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getConsumers().size(), 2);
+ assertEquals(((PatternMultiTopicsConsumerImpl<byte[]>) consumer).getPartitionedTopics().size(), 1);
+ });
// 8. produce data to topic2, verify should receive all the message
for (int i = 0; i < totalMessages; i++) {
@@ -1179,10 +1215,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
.subscribe();
// Wait topic list watcher creation.
- Awaitility.await().untilAsserted(() -> {
- CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture");
- assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally());
- });
+ waitTopicListWatcherCreation(consumer);
assertTrue(consumer instanceof PatternMultiTopicsConsumerImpl);
PatternMultiTopicsConsumerImpl<String> consumerImpl = (PatternMultiTopicsConsumerImpl<String>) consumer;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index 184b210e2b0..a269ac1b570 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -191,11 +191,14 @@ public class ClientCnx extends PulsarHandler {
@Getter
protected AuthenticationDataProvider authenticationDataProvider;
private TransactionBufferHandler transactionBufferHandler;
+ @Getter
private boolean supportsTopicWatchers;
@Getter
private boolean supportsGetPartitionedMetadataWithoutAutoCreation;
@Getter
private boolean brokerSupportsReplDedupByLidAndEid;
+ @Getter
+ private boolean supportsTopicWatcherReconcile;
/** Idle stat. **/
@Getter
@@ -411,6 +414,8 @@ public class ClientCnx extends PulsarHandler {
&& connected.getFeatureFlags().isSupportsGetPartitionedMetadataWithoutAutoCreation();
brokerSupportsReplDedupByLidAndEid =
connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsReplDedupByLidAndEid();
+ supportsTopicWatcherReconcile =
+ connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatcherReconcile();
// set remote protocol version to the correct version before we complete the connection future
setRemoteEndpointProtocolVersion(connected.getProtocolVersion());
@@ -1224,6 +1229,16 @@ public class ClientCnx extends PulsarHandler {
RequestType.Command, true);
}
+ /**
+ * Create and send a WatchTopicList request including the topics-hash.
+ * Delegates to the existing BaseCommand-based method after building the command.
+ */
+ public CompletableFuture<CommandWatchTopicListSuccess> newWatchTopicList(
+ long requestId, long watcherId, String namespace, String topicsPattern, String topicsHash) {
+ BaseCommand cmd = Commands.newWatchTopicList(requestId, watcherId, namespace, topicsPattern, topicsHash);
+ return newWatchTopicList(cmd, requestId);
+ }
+
public CompletableFuture<CommandSuccess> newWatchTopicListClose(
BaseCommand commandWatchTopicListClose, long requestId) {
return sendRequestAndHandleTimeout(
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
index d6eba6463a0..49b0add21e7 100644
--- 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
@@ -26,6 +26,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
/**
* Used to make all tasks that will modify subscriptions will be executed one by one, and skip the unnecessary updating.
@@ -47,15 +48,76 @@ import org.apache.commons.lang3.tuple.Pair;
@Slf4j
@SuppressFBWarnings("EI_EXPOSE_REP2")
public class PatternConsumerUpdateQueue {
+ private final LinkedBlockingQueue<UpdateTask> pendingTasks;
- private static final Pair<UpdateSubscriptionType, Collection<String>> RECHECK_OP =
- Pair.of(UpdateSubscriptionType.RECHECK, null);
+ private final PatternMultiTopicsConsumerImpl<?> patternConsumer;
- private final LinkedBlockingQueue<Pair<UpdateSubscriptionType, Collection<String>>> pendingTasks;
+ private final PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener;
- private final PatternMultiTopicsConsumerImpl patternConsumer;
+ static class UpdateTask {
+ private final UpdateSubscriptionType type;
- private final PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener;
+ UpdateTask(UpdateSubscriptionType type) {
+ this.type = type;
+ }
+ }
+
+ static class RecheckTask extends UpdateTask {
+ public static final RecheckTask INSTANCE = new RecheckTask();
+ private RecheckTask() {
+ super(UpdateSubscriptionType.RECHECK);
+ }
+
+ @Override
+ public String toString() {
+ return "RecheckTask";
+ }
+ }
+
+ static class InitTask extends UpdateTask {
+ public static final InitTask INSTANCE = new InitTask();
+ private InitTask() {
+ super(UpdateSubscriptionType.CONSUMER_INIT);
+ }
+
+ @Override
+ public String toString() {
+ return "InitTask";
+ }
+ }
+
+ static class TopicsAddedOrRemovedTask extends UpdateTask {
+ private final Collection<String> addedTopics;
+ private final Collection<String> removedTopics;
+ private final String topicsHash;
+
+ public TopicsAddedOrRemovedTask(Collection<String> addedTopics, Collection<String> removedTopics,
+ String topicsHash) {
+ super(UpdateSubscriptionType.TOPICS_CHANGED);
+ this.addedTopics = addedTopics;
+ this.removedTopics = removedTopics;
+ this.topicsHash = topicsHash;
+ }
+
+ @Override
+ public String toString() {
+ return "TopicsAddedOrRemovedTask{" + "addedTopics=" + addedTopics + ", removedTopics=" + removedTopics
+ + ", topicsHash='" + topicsHash + '\'' + '}';
+ }
+ }
+
+ static class WatchTopicListSuccessTask extends UpdateTask {
+ private final CommandWatchTopicListSuccess response;
+ private final String localStateTopicsHash;
+ private final int epoch;
+
+ WatchTopicListSuccessTask(CommandWatchTopicListSuccess response, String localStateTopicsHash, int epoch) {
+ super(UpdateSubscriptionType.WATCH_TOPIC_LIST_SUCCESS);
+ this.response = response;
+ this.localStateTopicsHash = localStateTopicsHash;
+ this.epoch = epoch;
+ }
+ }
/**
* Whether there is a task is in progress, this variable is used to confirm whether a next-task triggering is
@@ -75,44 +137,40 @@ public class PatternConsumerUpdateQueue {
private boolean closed;
- public PatternConsumerUpdateQueue(PatternMultiTopicsConsumerImpl patternConsumer) {
+ public PatternConsumerUpdateQueue(PatternMultiTopicsConsumerImpl<?> patternConsumer) {
this(patternConsumer, patternConsumer.topicsChangeListener);
}
/** This constructor is only for test. **/
@VisibleForTesting
- public PatternConsumerUpdateQueue(PatternMultiTopicsConsumerImpl patternConsumer,
+ @SuppressWarnings("this-escape")
+ 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));
+ doAppend(InitTask.INSTANCE);
}
- synchronized void appendTopicsAddedOp(Collection<String> topics) {
- if (topics == null || topics.isEmpty()) {
- return;
- }
- doAppend(Pair.of(UpdateSubscriptionType.TOPICS_ADDED, topics));
+ synchronized void appendTopicsChangedOp(Collection<String> addedTopics, Collection<String> deletedTopics,
+ String topicsHash) {
+ doAppend(new TopicsAddedOrRemovedTask(addedTopics, deletedTopics, topicsHash));
}
- synchronized void appendTopicsRemovedOp(Collection<String> topics) {
- if (topics == null || topics.isEmpty()) {
- return;
- }
- doAppend(Pair.of(UpdateSubscriptionType.TOPICS_REMOVED, topics));
+ synchronized void appendRecheckOp() {
+ doAppend(RecheckTask.INSTANCE);
}
- synchronized void appendRecheckOp() {
- doAppend(RECHECK_OP);
+ synchronized void appendWatchTopicListSuccessOp(CommandWatchTopicListSuccess response, String localStateTopicsHash,
+ int epoch) {
+ doAppend(new WatchTopicListSuccessTask(response, localStateTopicsHash, epoch));
}
- synchronized void doAppend(Pair<UpdateSubscriptionType, Collection<String>> task) {
+ synchronized void doAppend(UpdateTask task) {
if (log.isDebugEnabled()) {
- log.debug("Pattern consumer [{}] try to append task. {} {}", patternConsumer.getSubscription(),
- task.getLeft(), task.getRight() == null ? "" : task.getRight());
+ log.debug("Pattern consumer [{}] try to append task. {}", patternConsumer.getSubscription(), task);
}
// Once there is a recheck task in queue, it means other tasks can be skipped.
if (recheckTaskInQueue) {
@@ -120,13 +178,13 @@ public class PatternConsumerUpdateQueue {
}
// Once there are too many tasks in queue, compress them as a recheck task.
- if (pendingTasks.size() >= 30 && !task.getLeft().equals(UpdateSubscriptionType.RECHECK)) {
+ if (pendingTasks.size() >= 30 && task.type != UpdateSubscriptionType.RECHECK) {
appendRecheckOp();
return;
}
pendingTasks.add(task);
- if (task.getLeft().equals(UpdateSubscriptionType.RECHECK)) {
+ if (task.type == UpdateSubscriptionType.RECHECK) {
recheckTaskInQueue = true;
}
@@ -141,7 +199,7 @@ public class PatternConsumerUpdateQueue {
return;
}
- final Pair<UpdateSubscriptionType, Collection<String>> task = pendingTasks.poll();
+ final UpdateTask task = pendingTasks.poll();
// No pending task.
if (task == null) {
@@ -150,14 +208,14 @@ public class PatternConsumerUpdateQueue {
}
// If there is a recheck task in queue, skip others and only call the recheck task.
- if (recheckTaskInQueue && !task.getLeft().equals(UpdateSubscriptionType.RECHECK)) {
+ if (recheckTaskInQueue && task.type != UpdateSubscriptionType.RECHECK) {
triggerNextTask();
return;
}
// Execute pending task.
CompletableFuture<Void> newTaskFuture = null;
- switch (task.getLeft()) {
+ switch (task.type) {
case CONSUMER_INIT: {
newTaskFuture = patternConsumer.getSubscribeFuture().thenAccept(__ -> {}).exceptionally(ex -> {
// If the subscribe future was failed, the consumer will be closed.
@@ -173,12 +231,32 @@ public class PatternConsumerUpdateQueue {
});
break;
}
- case TOPICS_ADDED: {
- newTaskFuture = topicsChangeListener.onTopicsAdded(task.getRight());
- break;
- }
- case TOPICS_REMOVED: {
- newTaskFuture = topicsChangeListener.onTopicsRemoved(task.getRight());
+ case TOPICS_CHANGED: {
+ TopicsAddedOrRemovedTask topicsAddedOrRemovedTask = (TopicsAddedOrRemovedTask) task;
+ newTaskFuture = topicsChangeListener.onTopicsRemoved(topicsAddedOrRemovedTask.removedTopics)
+ .thenCompose(__ ->
+ topicsChangeListener.onTopicsAdded(topicsAddedOrRemovedTask.addedTopics))
+ .thenRun(() -> {
+ if (!patternConsumer.supportsTopicListWatcherReconcile()) {
+ // Ignore the topics hash until topic-list watcher reconciliation is supported.
+ // Broker-side state can be stale, which would trigger unnecessary reconciliation.
+ // The client will reconcile later when it fetches the topic list after the next
+ // patternAutoDiscoveryPeriod interval.
+ // Brokers that support watcher reconciliation also refresh broker-side state
+ // when reconciliation is requested.
+ // Older brokers have known topic-listing bugs (issue 25192: system topics included),
+ // so their hash is not reliable anyway.
+ return;
+ }
+ String localHash = patternConsumer.getLocalStateTopicsHash();
+ String brokerHash = topicsAddedOrRemovedTask.topicsHash;
+ if (brokerHash != null && brokerHash.length() > 0 && !brokerHash.equals(localHash)) {
+ log.info("[{}][{}] Hash mismatch detected (local: {}, broker: {}). Triggering "
+ + "reconciliation.", patternConsumer.getPattern().inputPattern(),
+ patternConsumer.getSubscription(), localHash, brokerHash);
+ appendRecheckOp();
+ }
+ });
break;
}
case RECHECK: {
@@ -187,20 +265,24 @@ public class PatternConsumerUpdateQueue {
newTaskFuture = patternConsumer.recheckTopicsChange();
break;
}
+ case WATCH_TOPIC_LIST_SUCCESS: {
+ WatchTopicListSuccessTask watchTopicListSuccessTask = (WatchTopicListSuccessTask) task;
+ newTaskFuture = patternConsumer.handleWatchTopicListSuccess(watchTopicListSuccessTask.response,
+ watchTopicListSuccessTask.localStateTopicsHash, watchTopicListSuccessTask.epoch);
+ 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());
+ log.debug("Pattern consumer [{}] starting task. {}", patternConsumer.getSubscription(), task);
}
// Trigger next pending task.
- taskInProgress = Pair.of(task.getLeft(), newTaskFuture);
+ taskInProgress = Pair.of(task.type, newTaskFuture);
newTaskFuture.thenAccept(ignore -> {
if (log.isDebugEnabled()) {
- log.debug("Pattern consumer [{}] task finished. {} {} ", patternConsumer.getSubscription(),
- task.getLeft(), task.getRight() == null ? "" : task.getRight());
+ log.debug("Pattern consumer [{}] task finished. {}", patternConsumer.getSubscription(), task);
}
triggerNextTask();
}).exceptionally(ex -> {
@@ -209,8 +291,8 @@ public class PatternConsumerUpdateQueue {
* - 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);
+ log.error("Pattern consumer [{}] task finished. {}. But it failed", patternConsumer.getSubscription(),
+ task, ex);
// Skip if there is already a recheck task in queue.
synchronized (PatternConsumerUpdateQueue.this) {
if (recheckTaskInQueue || PatternConsumerUpdateQueue.this.closed) {
@@ -242,13 +324,13 @@ public class PatternConsumerUpdateQueue {
}
private enum UpdateSubscriptionType {
- /** A marker that indicates the consumer's subscribe task.**/
+ /** 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,
+ /** Triggered by topic list watcher when topics changed. **/
+ TOPICS_CHANGED,
/** A fully check for pattern consumer. **/
- RECHECK;
+ RECHECK,
+ /** Handle initial watch topic list success response. **/
+ WATCH_TOPIC_LIST_SUCCESS;
}
}
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 8a3798a670c..3da14637a1a 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
@@ -33,11 +33,16 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import lombok.AccessLevel;
+import lombok.Getter;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.lookup.GetTopicsResult;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
@@ -52,15 +57,14 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
private final TopicsPattern topicsPattern;
final TopicsChangedListener topicsChangeListener;
private final Mode subscriptionMode;
- private final TopicListWatcher topicListWatcher;
+ @Getter(value = AccessLevel.PROTECTED, onMethod_ = @VisibleForTesting)
+ private volatile TopicListWatcher topicListWatcher;
+ @Getter(onMethod_ = @VisibleForTesting)
private final CompletableFuture<TopicListWatcher> watcherFuture = new CompletableFuture<>();
protected NamespaceName namespaceName;
private final AtomicInteger recheckPatternEpoch = new AtomicInteger();
- // If recheckPatternTimeout is not null, it means the broker's topic watcher is disabled.
- // The client need falls back to the polling model.
private volatile Timeout recheckPatternTimeout = null;
- private volatile String topicsHash;
private PatternConsumerUpdateQueue updateTaskQueue;
private volatile boolean closed = false;
@@ -69,7 +73,6 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
* @param topicsPattern The regexp for the topic name(not contains partition suffix).
*/
public PatternMultiTopicsConsumerImpl(TopicsPattern topicsPattern,
- String topicsHash,
PulsarClientImpl client,
ConsumerConfigurationData<T> conf,
ExecutorProvider executorProvider,
@@ -80,37 +83,45 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
super(client, conf, executorProvider, subscribeFuture, schema, interceptors,
false /* createTopicIfDoesNotExist */);
this.topicsPattern = topicsPattern;
- this.topicsHash = topicsHash;
this.subscriptionMode = subscriptionMode;
this.namespaceName = topicsPattern.namespace();
-
this.topicsChangeListener = new PatternTopicsChangedListener();
this.updateTaskQueue = new PatternConsumerUpdateQueue(this);
if (subscriptionMode == Mode.PERSISTENT) {
- long watcherId = client.newTopicListWatcherId();
- topicListWatcher = new TopicListWatcher(updateTaskQueue, client, topicsPattern, watcherId,
- namespaceName, topicsHash, watcherFuture, () -> recheckTopicsChangeAfterReconnect());
- watcherFuture
- .exceptionally(ex -> {
- if (closed) {
- log.warn("Pattern consumer [{}] was closed while creating topic list watcher",
- conf.getSubscriptionName(), ex);
- } else {
- log.warn(
- "Pattern consumer [{}] unable to create topic list watcher. Falling back to only polling"
- + " for new topics", conf.getSubscriptionName(), ex);
- this.recheckPatternTimeout = client.timer()
- .newTimeout(this, Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS);
- }
- return null;
- });
+ subscribeFuture.whenComplete((__, exception) -> {
+ if (!closed && exception == null) {
+ long watcherId = client.newTopicListWatcherId();
+ topicListWatcher = new TopicListWatcher(updateTaskQueue, client, topicsPattern, watcherId,
+ namespaceName, this::getLocalStateTopicsHash, watcherFuture,
+ this::getNextRecheckPatternEpoch);
+ watcherFuture.whenComplete((watcher, ex) -> {
+ if (closed) {
+ log.warn("Pattern consumer [{}] was closed while creating topic list watcher",
+ conf.getSubscriptionName(), ex);
+ } else if (ex != null) {
+ if (ex instanceof PulsarClientException.NotAllowedException) {
+ // create info message when topic watchers aren't supported
+ log.info("Pattern consumer [{}] unable to create topic list watcher. {}",
+ conf.getSubscriptionName(), ex.getMessage());
+ } else {
+ log.warn("Pattern consumer [{}] unable to create topic list watcher.",
+ conf.getSubscriptionName(), ex);
+ }
+ }
+ scheduleRecheckTopics();
+ });
+ }
+ });
} else {
log.debug("Pattern consumer [{}] not creating topic list watcher for subscription mode {}",
conf.getSubscriptionName(), subscriptionMode);
topicListWatcher = null;
watcherFuture.complete(null);
- this.recheckPatternTimeout = client.timer().newTimeout(
- this, Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS);
+ subscribeFuture.whenComplete((__, ex) -> {
+ if (!closed && ex == null) {
+ scheduleRecheckTopics();
+ }
+ });
}
}
@@ -136,46 +147,123 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
}
CompletableFuture<Void> recheckTopicsChange() {
- String pattern = topicsPattern.inputPattern();
- final int epoch = recheckPatternEpoch.incrementAndGet();
- return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash)
- .thenCompose(getTopicsResult -> {
- // If "recheckTopicsChange" has been called more than one times, only make the last one take affects.
- // Use "synchronized (recheckPatternTaskBackoff)" instead of
- // `synchronized(PatternMultiTopicsConsumerImpl.this)` to avoid locking in a wider range.
- synchronized (PatternMultiTopicsConsumerImpl.this) {
- if (recheckPatternEpoch.get() > epoch) {
- return CompletableFuture.completedFuture(null);
- }
- if (log.isDebugEnabled()) {
- 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 int epoch = getNextRecheckPatternEpoch();
- final List<String> oldTopics = new ArrayList<>(getPartitions());
- return updateSubscriptions(topicsPattern, this::setTopicsHash, getTopicsResult,
- topicsChangeListener, oldTopics, subscription);
- }
- }).thenAccept(__ -> {
- if (recheckPatternTimeout != null) {
- this.recheckPatternTimeout = client.timer().newTimeout(
- this, Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS);
+ CompletableFuture<Void> recheckFuture;
+ // Prefer watcher-based reconcile when a watcher exists and is connected. Fallback to lookup if watcher
+ // is not available or the watcher-based request fails.
+ if (supportsTopicListWatcherReconcile()) {
+ String localStateTopicsHash = getLocalStateTopicsHash();
+ recheckFuture = topicListWatcher.reconcile(localStateTopicsHash).thenCompose(response -> {
+ return handleWatchTopicListSuccess(response, localStateTopicsHash, epoch);
+ }).handle((res, ex) -> {
+ if (ex != null) {
+ // watcher-based reconcile failed -> fall back to lookup-based recheck
+ return doLookupBasedRecheck(epoch);
+ } else {
+ // watcher-based reconcile completed successfully
+ return CompletableFuture.<Void>completedFuture(null);
}
- });
+ }).thenCompose(Function.identity());
+ } else {
+ // Fallback: perform the existing lookup-based recheck
+ recheckFuture = doLookupBasedRecheck(epoch);
+ }
+
+ return recheckFuture.handle((__, ex) -> {
+ if (ex != null) {
+ log.info("[{}][{}] Pattern consumer failed to recheck topics changes: {}",
+ getPattern().inputPattern(), getSubscription(), ex.getMessage());
+ }
+ scheduleRecheckTopics();
+ return null;
+ });
+ }
+
+ int getNextRecheckPatternEpoch() {
+ return recheckPatternEpoch.incrementAndGet();
+ }
+
+ CompletableFuture<Void> handleWatchTopicListSuccess(CommandWatchTopicListSuccess response,
+ String localStateTopicsHash, int epoch) {
+ synchronized (PatternMultiTopicsConsumerImpl.this) {
+ if (recheckPatternEpoch.get() > epoch) {
+ return CompletableFuture.completedFuture(null);
+ }
+ // Build a GetTopicsResult-like object from the watch response
+ // so we can reuse updateSubscriptions
+ final List<String> topics = (response != null)
+ ? response.getTopicsList()
+ : Collections.emptyList();
+ final String hash = (response != null && response.hasTopicsHash())
+ ? response.getTopicsHash()
+ : null;
+ final boolean changed = !localStateTopicsHash.equals(hash);
+ final GetTopicsResult getTopicsResult =
+ new GetTopicsResult(topics, hash, true, changed);
+
+ final List<String> oldTopics = new ArrayList<>(getPartitions());
+ return updateSubscriptions(topicsPattern, getTopicsResult, topicsChangeListener, oldTopics,
+ subscription);
+ }
+ }
+
+ boolean supportsTopicListWatcherReconcile() {
+ return topicListWatcher != null && topicListWatcher.supportsReconcile() && watcherFuture.isDone()
+ && !watcherFuture.isCompletedExceptionally() && topicListWatcher.isConnected();
+ }
+
+ private synchronized void scheduleRecheckTopics() {
+ if (!closed) {
+ // cancel previous timeout if it exists
+ Timeout oldTimeout = this.recheckPatternTimeout;
+ if (oldTimeout != null) {
+ // cancel is a no-op if the timeout has already been executed or cancelled
+ oldTimeout.cancel();
+ }
+ this.recheckPatternTimeout = client.timer().newTimeout(this,
+ Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS);
+ }
+ }
+
+ private CompletableFuture<Void> doLookupBasedRecheck(final int epoch) {
+ final String pattern = topicsPattern.inputPattern();
+ return client.getLookup()
+ .getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, getLocalStateTopicsHash())
+ .thenCompose(getTopicsResult -> {
+ // If "recheckTopicsChange" has been called more than one times, only make the last one take
+ // affects.
+ // Use "synchronized (recheckPatternTaskBackoff)" instead of
+ // `synchronized(PatternMultiTopicsConsumerImpl.this)` to avoid locking in a wider range.
+ synchronized (PatternMultiTopicsConsumerImpl.this) {
+ if (recheckPatternEpoch.get() > epoch) {
+ return CompletableFuture.completedFuture(null);
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Pattern consumer [{}] get topics under namespace {}, "
+ + "topics.size: {}, "
+ + "topicsHash: {}, filtered: {}",
+ 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<>(getPartitions());
+ return updateSubscriptions(topicsPattern, getTopicsResult, topicsChangeListener, oldTopics,
+ subscription);
+ }
+ });
}
static CompletableFuture<Void> updateSubscriptions(TopicsPattern topicsPattern,
- java.util.function.Consumer<String> topicsHashSetter,
GetTopicsResult getTopicsResult,
TopicsChangedListener topicsChangedListener,
List<String> oldTopics,
String subscriptionForLog) {
- topicsHashSetter.accept(getTopicsResult.getTopicsHash());
if (!getTopicsResult.isChanged()) {
return CompletableFuture.completedFuture(null);
}
@@ -203,11 +291,6 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
return this.topicsPattern;
}
- @VisibleForTesting
- void setTopicsHash(String topicsHash) {
- this.topicsHash = topicsHash;
- }
-
interface TopicsChangedListener {
/***
* unsubscribe and delete {@link ConsumerImpl} in the {@link MultiTopicsConsumerImpl#consumers} map in
@@ -414,6 +497,15 @@ public class PatternMultiTopicsConsumerImpl<T> extends MultiTopicsConsumerImpl<T
return recheckPatternTimeout;
}
+ /**
+ * Get the current topics hash calculated from the pattern consumer's topic list.
+ * This is used to validate incremental updates against the broker's hash.
+ */
+ @VisibleForTesting
+ String getLocalStateTopicsHash() {
+ return TopicList.calculateHash(getPartitions());
+ }
+
protected void handleSubscribeOneTopicError(String topicName,
Throwable error,
CompletableFuture<Void> subscribeFuture) {
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 605757c230c..2bcf9779a03 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
@@ -50,6 +50,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import lombok.Builder;
import lombok.Getter;
import org.apache.commons.lang3.tuple.Pair;
@@ -684,9 +685,14 @@ public class PulsarClientImpl implements PulsarClient {
conf.getSubscriptionName(), namespaceName, topicName));
}
- List<String> topicsList = getTopicsResult.getTopics();
+ List<String> topicsList;
if (!getTopicsResult.isFiltered()) {
topicsList = TopicList.filterTopics(getTopicsResult.getTopics(), pattern);
+ } else {
+ // deduplicate java.lang.String instances using TopicName's cache
+ topicsList = getTopicsResult.getTopics().stream()
+ .map(TopicName::get).map(TopicName::toString)
+ .collect(Collectors.toList());
}
conf.getTopicNames().addAll(topicsList);
@@ -698,7 +704,6 @@ public class PulsarClientImpl implements PulsarClient {
// Pattern consumer has his unique check mechanism, so do not need the feature "autoUpdatePartitions".
conf.setAutoUpdatePartitions(false);
ConsumerBase<T> consumer = new PatternMultiTopicsConsumerImpl<>(pattern,
- getTopicsResult.getTopicsHash(),
PulsarClientImpl.this,
conf,
externalExecutorProvider,
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 49cf64656fe..0c854a068a2 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
@@ -18,14 +18,20 @@
*/
package org.apache.pulsar.client.impl;
+import com.google.common.annotations.VisibleForTesting;
import io.netty.channel.ChannelHandlerContext;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.IntSupplier;
+import java.util.function.Supplier;
+import lombok.AccessLevel;
+import lombok.Getter;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.protocol.Commands;
@@ -46,27 +52,27 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
private final String name;
private final ConnectionHandler connectionHandler;
private final TopicsPattern topicsPattern;
+ @Getter(value = AccessLevel.PROTECTED, onMethod_ = @VisibleForTesting)
private final long watcherId;
private volatile long createWatcherDeadline = 0;
private final NamespaceName namespace;
- // TODO maintain the value based on updates from broker and warn the user if inconsistent with hash from polling
- private String topicsHash;
+ // maintain the value based on updates from broker and allow external components to query it
+ private final Supplier<String> localStateTopicsHashSupplier;
private final CompletableFuture<TopicListWatcher> watcherFuture;
+ private final IntSupplier nextRecheckPatternEpochSupplier;
private final AtomicInteger previousExceptionCount = new AtomicInteger();
private final AtomicReference<ClientCnx> clientCnxUsedForWatcherRegistration = new AtomicReference<>();
- private final Runnable recheckTopicsChangeAfterReconnect;
-
/***
* @param topicsPattern The regexp for the topic name(not contains partition suffix).
*/
public TopicListWatcher(PatternConsumerUpdateQueue patternConsumerUpdateQueue,
PulsarClientImpl client, TopicsPattern topicsPattern, long watcherId,
- NamespaceName namespace, String topicsHash,
+ NamespaceName namespace, Supplier<String> localStateTopicsHashSupplier,
CompletableFuture<TopicListWatcher> watcherFuture,
- Runnable recheckTopicsChangeAfterReconnect) {
+ IntSupplier nextRecheckPatternEpochSupplier) {
super(client, topicsPattern.topicLookupNameForTopicListWatcherPlacement());
this.patternConsumerUpdateQueue = patternConsumerUpdateQueue;
this.name = "Watcher(" + topicsPattern + ")";
@@ -81,9 +87,9 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
this.topicsPattern = topicsPattern;
this.watcherId = watcherId;
this.namespace = namespace;
- this.topicsHash = topicsHash;
+ this.localStateTopicsHashSupplier = localStateTopicsHashSupplier;
this.watcherFuture = watcherFuture;
- this.recheckTopicsChangeAfterReconnect = recheckTopicsChangeAfterReconnect;
+ this.nextRecheckPatternEpochSupplier = nextRecheckPatternEpochSupplier;
connectionHandler.grabCnx();
}
@@ -108,6 +114,15 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
@Override
public CompletableFuture<Void> connectionOpened(ClientCnx cnx) {
+ if (!cnx.isSupportsTopicWatchers()) {
+ watcherFuture.completeExceptionally(new PulsarClientException.NotAllowedException(
+ "Broker does not allow broker side pattern evaluation."));
+ setState(State.Closed);
+ deregisterFromClientCnx();
+ connectionClosed(cnx);
+ return CompletableFuture.completedFuture(null);
+ }
+
previousExceptionCount.set(0);
State state = getState();
@@ -130,11 +145,10 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
// synchronized this, because redeliverUnAckMessage eliminate the epoch inconsistency between them
synchronized (this) {
setClientCnx(cnx);
- BaseCommand watchRequest = Commands.newWatchTopicList(requestId, watcherId, namespace.toString(),
- topicsPattern.inputPattern(), topicsHash);
-
- cnx.newWatchTopicList(watchRequest, requestId)
-
+ String localStateTopicsHash = localStateTopicsHashSupplier.get();
+ int epoch = nextRecheckPatternEpochSupplier.getAsInt();
+ cnx.newWatchTopicList(requestId, watcherId, namespace.toString(), topicsPattern.inputPattern(),
+ localStateTopicsHash)
.thenAccept(response -> {
synchronized (TopicListWatcher.this) {
if (!changeToReadyState()) {
@@ -142,14 +156,15 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
// drops the watcher on its side
setState(State.Closed);
deregisterFromClientCnx();
+ log.warn("[{}] Watcher was closed while reconnecting, closing the connection to {}.",
+ topic, cnx.channel().remoteAddress());
cnx.channel().close();
future.complete(null);
return;
}
}
this.connectionHandler.resetBackoff();
-
- recheckTopicsChangeAfterReconnect.run();
+ patternConsumerUpdateQueue.appendWatchTopicListSuccessOp(response, localStateTopicsHash, epoch);
watcherFuture.complete(this);
future.complete(null);
}).exceptionally((e) -> {
@@ -280,7 +295,33 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
}
public void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate update) {
- patternConsumerUpdateQueue.appendTopicsRemovedOp(update.getDeletedTopicsList());
- patternConsumerUpdateQueue.appendTopicsAddedOp(update.getNewTopicsList());
+ if (update == null) {
+ return;
+ }
+ patternConsumerUpdateQueue.appendTopicsChangedOp(update.getNewTopicsList(), update.getDeletedTopicsList(),
+ update.hasTopicsHash() ? update.getTopicsHash() : "");
+ }
+
+ /**
+ * Perform a single reconciliation request using the existing watcher id and the watcher's last-known topics hash.
+ * This will send a WatchTopicList request including the topics-hash to the broker. If the watcher is not connected,
+ * the returned future will be completed exceptionally.
+ */
+ public CompletableFuture<CommandWatchTopicListSuccess> reconcile(String localStateTopicsHash) {
+ ClientCnx c = cnx();
+ if (c == null || !isConnected()) {
+ CompletableFuture<CommandWatchTopicListSuccess> f = new CompletableFuture<>();
+ f.completeExceptionally(new IllegalStateException("Watcher is not connected"));
+ return f;
+ }
+ long requestId = client.newRequestId();
+ // Use the convenience ClientCnx overload that accepts a topicsHash
+ return c.newWatchTopicList(requestId, watcherId, namespace.toString(),
+ topicsPattern.inputPattern(), localStateTopicsHash);
+ }
+
+ public boolean supportsReconcile() {
+ ClientCnx cnx = cnx();
+ return cnx != null && cnx.isSupportsTopicWatcherReconcile();
}
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
index e966b00b4a2..e4ab5da96f6 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
@@ -62,7 +62,7 @@ class ClientTestFixtures {
// return createPulsarClientMock(mock(ExecutorService.class));
// }
- static <T> PulsarClientImpl createPulsarClientMock(ExecutorProvider executorProvider,
+ public static <T> PulsarClientImpl createPulsarClientMock(ExecutorProvider executorProvider,
ExecutorService internalExecutorService) {
PulsarClientImpl clientMock = mock(PulsarClientImpl.class, Mockito.RETURNS_DEEP_STUBS);
@@ -82,11 +82,19 @@ class ClientTestFixtures {
static <T> PulsarClientImpl createPulsarClientMockWithMockedClientCnx(
ExecutorProvider executorProvider,
ExecutorService internalExecutorService) {
- return mockClientCnx(createPulsarClientMock(executorProvider, internalExecutorService));
+ return createPulsarClientMockWithMockedClientCnx(executorProvider, internalExecutorService, mockClientCnx());
}
- static PulsarClientImpl mockClientCnx(PulsarClientImpl clientMock) {
- ClientCnx clientCnxMock = mockClientCnx();
+ static <T> PulsarClientImpl createPulsarClientMockWithMockedClientCnx(
+ ExecutorProvider executorProvider,
+ ExecutorService internalExecutorService,
+ ClientCnx clientCnxMock) {
+ return clientMockWithClientCnxMock(createPulsarClientMock(executorProvider, internalExecutorService),
+ clientCnxMock
+ );
+ }
+
+ static PulsarClientImpl clientMockWithClientCnxMock(PulsarClientImpl clientMock, ClientCnx clientCnxMock) {
when(clientMock.getConnection(any())).thenReturn(CompletableFuture.completedFuture(clientCnxMock));
when(clientMock.getConnection(anyString())).thenReturn(CompletableFuture.completedFuture(clientCnxMock));
when(clientMock.getConnection(anyString(), anyInt()))
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
index 01f0be6a85e..e8294eb94c0 100644
--- 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
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.impl;
import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -28,6 +29,7 @@ import java.io.Closeable;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
+import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -35,6 +37,7 @@ import lombok.AllArgsConstructor;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.util.FutureUtil;
import org.awaitility.Awaitility;
+import org.mockito.stubbing.Answer;
import org.testng.annotations.Test;
@Test(groups = "utils")
@@ -65,6 +68,7 @@ public class PatternConsumerUpdateQueueTest {
} else {
when(patternConsumer.getSubscribeFuture()).thenReturn(CompletableFuture.completedFuture(null));
}
+ when(patternConsumer.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener =
mock(PatternMultiTopicsConsumerImpl.TopicsChangedListener.class);
@@ -72,11 +76,19 @@ public class PatternConsumerUpdateQueueTest {
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);
+ CompletableFuture<Void> ex = FutureUtil.failedFuture(new RuntimeException("Failed topics changed event"));
+ Answer answer = invocationOnMock -> {
+ Collection inputCollection = invocationOnMock.getArgument(0, Collection.class);
+ if (successTopics.containsAll(inputCollection)) {
+ return customizedPartialUpdateFuture;
+ } else if (errorTopics.containsAll(inputCollection)) {
+ return ex;
+ } else {
+ throw new RuntimeException("Unexpected topics changed event");
+ }
+ };
+ doAnswer(answer).when(topicsChangeListener).onTopicsAdded(anyCollection());
+ doAnswer(answer).when(topicsChangeListener).onTopicsRemoved(anyCollection());
}
PatternConsumerUpdateQueue queue = new PatternConsumerUpdateQueue(patternConsumer, topicsChangeListener);
@@ -104,14 +116,14 @@ public class PatternConsumerUpdateQueueTest {
public void testTopicsChangedEvents() {
QueueInstance instance = createInstance();
- Collection<String> topics = Arrays.asList("a");
+ Collection<String> addedTopics = Arrays.asList("a");
+ Collection<String> removedTopics = Arrays.asList("b");
for (int i = 0; i < 10; i++) {
- instance.queue.appendTopicsAddedOp(topics);
- instance.queue.appendTopicsRemovedOp(topics);
+ instance.queue.appendTopicsChangedOp(addedTopics, removedTopics, "");
}
Awaitility.await().untilAsserted(() -> {
- verify(instance.mockedListener, times(10)).onTopicsAdded(topics);
- verify(instance.mockedListener, times(10)).onTopicsRemoved(topics);
+ verify(instance.mockedListener, times(10)).onTopicsAdded(addedTopics);
+ verify(instance.mockedListener, times(10)).onTopicsRemoved(removedTopics);
});
// cleanup.
@@ -165,11 +177,11 @@ public class PatternConsumerUpdateQueueTest {
CompletableFuture<Void> consumerInitFuture = CompletableFuture.completedFuture(null);
QueueInstance instance = createInstance(recheckFuture, partialUpdateFuture, consumerInitFuture);
- Collection<String> topics = Arrays.asList("a");
+ Collection<String> addedTopics = Arrays.asList("a");
+ Collection<String> removedTopics = Arrays.asList("b");
for (int i = 0; i < 10; i++) {
instance.queue.appendRecheckOp();
- instance.queue.appendTopicsAddedOp(topics);
- instance.queue.appendTopicsRemovedOp(topics);
+ instance.queue.appendTopicsChangedOp(addedTopics, removedTopics, "");
}
recheckFuture.complete(null);
Awaitility.await().untilAsserted(() -> {
@@ -182,8 +194,8 @@ public class PatternConsumerUpdateQueueTest {
// "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);
+ verify(instance.mockedListener, times(0)).onTopicsAdded(addedTopics);
+ verify(instance.mockedListener, times(0)).onTopicsRemoved(removedTopics);
});
// cleanup.
@@ -193,20 +205,21 @@ public class PatternConsumerUpdateQueueTest {
@Test
public void testErrorTask() {
CompletableFuture<Void> immediatelyCompleteFuture = CompletableFuture.completedFuture(null);
- Collection<String> successTopics = Arrays.asList("a");
+ Collection<String> successTopics = Arrays.asList("a", "b");
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);
+ Collection<String> addedTopics = Arrays.asList("a");
+ Collection<String> removedTopics = Arrays.asList("b");
+
+ instance.queue.appendTopicsChangedOp(addedTopics, removedTopics, "");
+ instance.queue.appendTopicsChangedOp(errorTopics, List.of(), "");
+ instance.queue.appendTopicsChangedOp(addedTopics, removedTopics, "");
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(2)).onTopicsAdded(addedTopics);
+ verify(instance.mockedListener, times(2)).onTopicsRemoved(removedTopics);
verify(instance.mockedListener, times(1)).onTopicsAdded(errorTopics);
// After an error task will push a recheck task to offset.
verify(instance.mockedConsumer, times(1)).recheckTopicsChange();
@@ -225,18 +238,19 @@ public class PatternConsumerUpdateQueueTest {
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);
+ Collection<String> addedTopics = Arrays.asList("a");
+ Collection<String> removedTopics = Arrays.asList("b");
+
+ instance.queue.appendTopicsChangedOp(addedTopics, removedTopics, "");
+ instance.queue.appendTopicsChangedOp(errorTopics, List.of(), "");
+ instance.queue.appendTopicsChangedOp(addedTopics, removedTopics, "");
// 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(addedTopics);
+ verify(instance.mockedListener, times(0)).onTopicsRemoved(removedTopics);
verify(instance.mockedListener, times(0)).onTopicsAdded(errorTopics);
verify(instance.mockedConsumer, times(0)).recheckTopicsChange();
});
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 4bb541fbd5a..dd4bbff757a 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
@@ -18,43 +18,98 @@
*/
package org.apache.pulsar.client.impl;
-import static org.mockito.Mockito.any;
+import static org.apache.pulsar.client.impl.ClientTestFixtures.createDelayedCompletedFuture;
+import static org.apache.pulsar.client.impl.ClientTestFixtures.createPulsarClientMockWithMockedClientCnx;
+import static org.apache.pulsar.client.impl.ClientTestFixtures.mockClientCnx;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.Sets;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.TimerTask;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Deque;
+import java.util.HashSet;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
+import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
+import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.lookup.GetTopicsResult;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.common.topics.TopicList;
+import org.apache.pulsar.common.topics.TopicsPattern;
import org.apache.pulsar.common.topics.TopicsPatternFactory;
+import org.awaitility.Awaitility;
+import org.jspecify.annotations.NonNull;
+import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+/**
+ * Unit Tests of {@link PatternMultiTopicsConsumerImpl}.
+ */
public class PatternMultiTopicsConsumerImplTest {
-
private PatternMultiTopicsConsumerImpl.TopicsChangedListener mockListener;
-
- private Consumer<String> mockTopicsHashSetter;
-
+ private ExecutorProvider executorProvider;
+ private ExecutorService internalExecutor;
+ private PulsarClientImpl clientMock;
+ private ClientCnx cnx;
+ private Timer timer;
@BeforeMethod(alwaysRun = true)
public void setUp() {
+ executorProvider = new ExecutorProvider(1, "PatternMultiTopicsConsumerImplTest");
+ internalExecutor = Executors.newSingleThreadScheduledExecutor();
+ cnx = mockClientCnx();
+ clientMock = createPulsarClientMockWithMockedClientCnx(executorProvider, internalExecutor, cnx);
+ timer = new HashedWheelTimer();
+ when(clientMock.timer()).thenReturn(timer);
mockListener = mock(PatternMultiTopicsConsumerImpl.TopicsChangedListener.class);
when(mockListener.onTopicsAdded(any())).thenReturn(CompletableFuture.completedFuture(null));
when(mockListener.onTopicsRemoved(any())).thenReturn(CompletableFuture.completedFuture(null));
- mockTopicsHashSetter = mock(Consumer.class);
+ }
+ @AfterMethod(alwaysRun = true)
+ public void cleanUp() {
+ if (executorProvider != null) {
+ executorProvider.shutdownNow();
+ executorProvider = null;
+ }
+ if (internalExecutor != null) {
+ internalExecutor.shutdownNow();
+ internalExecutor = null;
+ }
+ if (timer != null) {
+ timer.stop();
+ }
}
@Test
public void testChangedUnfilteredResponse() {
PatternMultiTopicsConsumerImpl.updateSubscriptions(
TopicsPatternFactory.create(Pattern.compile("tenant/my-ns/name-.*")),
- mockTopicsHashSetter,
new GetTopicsResult(Arrays.asList(
"persistent://tenant/my-ns/name-1",
"persistent://tenant/my-ns/name-2",
@@ -66,14 +121,12 @@ public class PatternMultiTopicsConsumerImplTest {
"persistent://tenant/my-ns/name-1",
"persistent://tenant/my-ns/name-2"));
verify(mockListener).onTopicsRemoved(Collections.emptySet());
- verify(mockTopicsHashSetter).accept(null);
}
@Test
public void testChangedFilteredResponse() {
PatternMultiTopicsConsumerImpl.updateSubscriptions(
TopicsPatternFactory.create(Pattern.compile("tenant/my-ns/name-.*")),
- mockTopicsHashSetter,
new GetTopicsResult(Arrays.asList(
"persistent://tenant/my-ns/name-0",
"persistent://tenant/my-ns/name-1",
@@ -85,14 +138,12 @@ public class PatternMultiTopicsConsumerImplTest {
"persistent://tenant/my-ns/name-1",
"persistent://tenant/my-ns/name-2"));
verify(mockListener).onTopicsRemoved(Collections.emptySet());
- verify(mockTopicsHashSetter).accept("TOPICS_HASH");
}
@Test
public void testUnchangedResponse() {
PatternMultiTopicsConsumerImpl.updateSubscriptions(
TopicsPatternFactory.create(Pattern.compile("tenant/my-ns/name-.*")),
- mockTopicsHashSetter,
new GetTopicsResult(Arrays.asList(
"persistent://tenant/my-ns/name-0",
"persistent://tenant/my-ns/name-1",
@@ -102,6 +153,197 @@ public class PatternMultiTopicsConsumerImplTest {
Arrays.asList("persistent://tenant/my-ns/name-0"), "");
verify(mockListener, never()).onTopicsAdded(any());
verify(mockListener, never()).onTopicsRemoved(any());
- verify(mockTopicsHashSetter).accept("TOPICS_HASH");
+ }
+
+ @Test
+ public void testPatternSubscribeAndReconcileLoop() throws Exception {
+ TopicsPattern topicsPattern =
+ TopicsPatternFactory.create("persistent://tenant/namespace/.*", TopicsPattern.RegexImplementation.JDK);
+ ConsumerConfigurationData<byte[]> consumerConfData = createConsumerConfigurationData();
+ consumerConfData.setPatternAutoDiscoveryPeriod(1);
+
+ CopyOnWriteArrayList<String> topics = new CopyOnWriteArrayList<>();
+ topics.add("persistent://tenant/namespace/topic1");
+ doAnswer(invocationOnMock -> {
+ long requestId = invocationOnMock.getArgument(0);
+ long watcherId = invocationOnMock.getArgument(1);
+ String localHash = invocationOnMock.getArgument(4);
+ CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess();
+ success.setRequestId(requestId);
+ success.setWatcherId(watcherId);
+ List<String> topicsCopy = new ArrayList<>(topics);
+ String calculatedHash = TopicList.calculateHash(topicsCopy);
+ if (!localHash.equals(calculatedHash)) {
+ success.addAllTopics(topicsCopy);
+ }
+ success.setTopicsHash(calculatedHash);
+ return CompletableFuture.completedFuture(success);
+ }).when(cnx).newWatchTopicList(anyLong(), anyLong(), any(), any(), any());
+ doReturn(true).when(cnx).isSupportsTopicWatchers();
+ doReturn(true).when(cnx).isSupportsTopicWatcherReconcile();
+
+ PatternMultiTopicsConsumerImpl<byte[]> consumer =
+ createPatternMultiTopicsConsumer(consumerConfData, topicsPattern);
+ assertThat(consumer.subscribeFuture).succeedsWithin(Duration.ofSeconds(5));
+ assertThat(consumer.getWatcherFuture()).succeedsWithin(Duration.ofSeconds(5));
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic1");
+ });
+ topics.add("persistent://tenant/namespace/topic2");
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic1",
+ "persistent://tenant/namespace/topic2");
+ });
+ }
+
+ @Test
+ public void testPatternSubscribeWithoutWatcher() throws Exception {
+ TopicsPattern topicsPattern =
+ TopicsPatternFactory.create("persistent://tenant/namespace/.*", TopicsPattern.RegexImplementation.JDK);
+ ConsumerConfigurationData<byte[]> consumerConfData = createConsumerConfigurationData();
+ consumerConfData.setPatternAutoDiscoveryPeriod(1);
+
+ CopyOnWriteArrayList<String> topics = new CopyOnWriteArrayList<>();
+ topics.add("persistent://tenant/namespace/topic1");
+ LookupService mockLookup = clientMock.getLookup();
+ doAnswer(invocationOnMock -> {
+ String localHash = invocationOnMock.getArgument(3);
+ List<String> topicsCopy = new ArrayList<>(topics);
+ String calculatedHash = TopicList.calculateHash(topicsCopy);
+ boolean changed = false;
+ if (!localHash.equals(calculatedHash)) {
+ changed = true;
+ }
+ GetTopicsResult result = new GetTopicsResult(topicsCopy, calculatedHash, false, changed);
+ return CompletableFuture.completedFuture(result);
+ }).when(mockLookup).getTopicsUnderNamespace(any(), any(), any(), any());
+ doReturn(false).when(cnx).isSupportsTopicWatchers();
+
+ PatternMultiTopicsConsumerImpl<byte[]> consumer =
+ createPatternMultiTopicsConsumer(consumerConfData, topicsPattern);
+ assertThat(consumer.subscribeFuture).succeedsWithin(Duration.ofSeconds(5));
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic1");
+ });
+ topics.add("persistent://tenant/namespace/topic2");
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic1",
+ "persistent://tenant/namespace/topic2");
+ });
+ }
+
+ @Test
+ public void testPatternSubscribeAndHashHandlingWithChanges() throws Exception {
+ TopicsPattern topicsPattern =
+ TopicsPatternFactory.create("persistent://tenant/namespace/.*", TopicsPattern.RegexImplementation.JDK);
+ ConsumerConfigurationData<byte[]> consumerConfData = createConsumerConfigurationData();
+ consumerConfData.setPatternAutoDiscoveryPeriod(5);
+ Timer timer = mock(Timer.class);
+ when(clientMock.timer()).thenReturn(timer);
+ Deque<TimerTask> tasks = new ConcurrentLinkedDeque<>();
+ doAnswer(invocationOnMock -> {
+ TimerTask task = invocationOnMock.getArgument(0);
+ tasks.add(task);
+ return mock(Timeout.class);
+ }).when(timer).newTimeout(any(), anyLong(), any());
+ CopyOnWriteArrayList<String> topics = new CopyOnWriteArrayList<>();
+ topics.add("persistent://tenant/namespace/topic1");
+ consumerConfData.setTopicNames(new HashSet<>(topics));
+ AtomicInteger invocationCount = new AtomicInteger(0);
+ doAnswer(invocationOnMock -> {
+ invocationCount.incrementAndGet();
+ long requestId = invocationOnMock.getArgument(0);
+ long watcherId = invocationOnMock.getArgument(1);
+ String localHash = invocationOnMock.getArgument(4);
+ CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess();
+ success.setRequestId(requestId);
+ success.setWatcherId(watcherId);
+ List<String> topicsCopy = new ArrayList<>(topics);
+ String calculatedHash = TopicList.calculateHash(topicsCopy);
+ if (!localHash.equals(calculatedHash)) {
+ throw new RuntimeException("Assuming no changes");
+ }
+ success.setTopicsHash(calculatedHash);
+ return CompletableFuture.completedFuture(success);
+ }).when(cnx).newWatchTopicList(anyLong(), anyLong(), any(), any(), any());
+ doReturn(true).when(cnx).isSupportsTopicWatchers();
+ doReturn(true).when(cnx).isSupportsTopicWatcherReconcile();
+
+ PatternMultiTopicsConsumerImpl<byte[]> consumer =
+ createPatternMultiTopicsConsumer(consumerConfData, topicsPattern);
+ assertThat(consumer.subscribeFuture).succeedsWithin(Duration.ofSeconds(5));
+ assertThat(consumer.getWatcherFuture()).succeedsWithin(Duration.ofSeconds(5));
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic1");
+ });
+ runTimerTasks(tasks);
+ topics.add("persistent://tenant/namespace/topic2");
+ CommandWatchTopicUpdate update = new CommandWatchTopicUpdate();
+ TopicListWatcher topicListWatcher = consumer.getTopicListWatcher();
+ update.setWatcherId(topicListWatcher.getWatcherId());
+ update.addNewTopic("persistent://tenant/namespace/topic2");
+ update.setTopicsHash(TopicList.calculateHash(topics));
+ topicListWatcher.handleCommandWatchTopicUpdate(update);
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic1",
+ "persistent://tenant/namespace/topic2");
+ });
+ runTimerTasks(tasks);
+ runTimerTasks(tasks);
+ assertThat(invocationCount.get()).isEqualTo(4);
+ CommandWatchTopicUpdate update2 = new CommandWatchTopicUpdate();
+ update2.setWatcherId(topicListWatcher.getWatcherId());
+ topics.add("persistent://tenant/namespace/topic3");
+ update2.addNewTopic("persistent://tenant/namespace/topic3");
+ topics.add("persistent://tenant/namespace/topic4");
+ update2.addNewTopic("persistent://tenant/namespace/topic4");
+ topics.remove("persistent://tenant/namespace/topic1");
+ update2.addDeletedTopic("persistent://tenant/namespace/topic1");
+ update2.setTopicsHash(TopicList.calculateHash(topics));
+ topicListWatcher.handleCommandWatchTopicUpdate(update2);
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(consumer.getPartitions()).containsExactlyInAnyOrder("persistent://tenant/namespace/topic2",
+ "persistent://tenant/namespace/topic3", "persistent://tenant/namespace/topic4");
+ });
+ assertThat(invocationCount.get()).isEqualTo(4);
+ runTimerTasks(tasks);
+ assertThat(invocationCount.get()).isEqualTo(5);
+ }
+
+ private static void runTimerTasks(Deque<TimerTask> tasks) throws Exception {
+ // first drain the queue to a list to avoid an infinite loop
+ List<TimerTask> taskList = new ArrayList<>();
+ while (!tasks.isEmpty()) {
+ taskList.add(tasks.poll());
+ }
+ // now run the tasks
+ for (TimerTask task : taskList) {
+ task.run(mock(Timeout.class));
+ }
+ }
+
+ private PatternMultiTopicsConsumerImpl<byte[]> createPatternMultiTopicsConsumer(TopicsPattern topicsPattern) {
+ ConsumerConfigurationData<byte[]> consumerConfData = createConsumerConfigurationData();
+ return createPatternMultiTopicsConsumer(consumerConfData, topicsPattern);
+ }
+
+ private static @NonNull ConsumerConfigurationData<byte[]> createConsumerConfigurationData() {
+ ConsumerConfigurationData<byte[]> consumerConfData = new ConsumerConfigurationData<>();
+ consumerConfData.setSubscriptionName("subscriptionName");
+ consumerConfData.setAutoUpdatePartitionsIntervalSeconds(0);
+ return consumerConfData;
+ }
+
+ private PatternMultiTopicsConsumerImpl<byte[]> createPatternMultiTopicsConsumer(
+ ConsumerConfigurationData<byte[]> consumerConfData, TopicsPattern topicsPattern) {
+ int completionDelayMillis = 100;
+ Schema<byte[]> schema = Schema.BYTES;
+ when(clientMock.getPartitionedTopicMetadata(any(), anyBoolean(), anyBoolean()))
+ .thenAnswer(invocation -> createDelayedCompletedFuture(
+ new PartitionedTopicMetadata(), completionDelayMillis));
+ PatternMultiTopicsConsumerImpl<byte[]> consumer = new PatternMultiTopicsConsumerImpl<byte[]>(
+ topicsPattern, clientMock, consumerConfData, executorProvider,
+ new CompletableFuture<>(), schema, CommandGetTopicsOfNamespace.Mode.PERSISTENT, null);
+ return consumer;
}
}
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 601045d17c1..9dccc7f7c1d 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
@@ -24,6 +24,7 @@ import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
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 static org.testng.Assert.assertTrue;
@@ -35,14 +36,13 @@ import java.util.concurrent.CompletableFuture;
import java.util.regex.Pattern;
import lombok.Cleanup;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.impl.PatternMultiTopicsConsumerImpl.TopicsChangedListener;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.topics.TopicsPatternFactory;
-import org.mockito.ArgumentCaptor;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -53,6 +53,7 @@ public class TopicListWatcherTest {
private PulsarClientImpl client;
private CompletableFuture<TopicListWatcher> watcherFuture;
private TopicsChangedListener listener;
+ private PatternMultiTopicsConsumerImpl<byte[]> patternConsumer;
@BeforeMethod(alwaysRun = true)
public void setup() {
@@ -74,17 +75,23 @@ public class TopicListWatcherTest {
when(connectionPool.getConnection(any(), any(), anyInt())).thenReturn(clientCnxFuture);
CompletableFuture<Void> completedFuture = CompletableFuture.completedFuture(null);
- PatternMultiTopicsConsumerImpl patternConsumer = mock(PatternMultiTopicsConsumerImpl.class);
- when(patternConsumer.getSubscribeFuture()).thenReturn(completedFuture);
+ patternConsumer = mock(PatternMultiTopicsConsumerImpl.class);
+ when(patternConsumer.getPattern()).thenReturn(TopicsPatternFactory.create(Pattern.compile(topic)));
+ when(patternConsumer.getPartitions()).thenReturn(Collections.singletonList("persistent://tenant/ns/topic11"));
+ CompletableFuture<Consumer<byte[]>> subscribeFuture = CompletableFuture.completedFuture(patternConsumer);
+ when(patternConsumer.getSubscribeFuture()).thenReturn(subscribeFuture);
when(patternConsumer.recheckTopicsChange()).thenReturn(completedFuture);
when(listener.onTopicsAdded(anyCollection())).thenReturn(completedFuture);
when(listener.onTopicsRemoved(anyCollection())).thenReturn(completedFuture);
+ when(patternConsumer.handleWatchTopicListSuccess(any(), any(), anyInt())).thenReturn(completedFuture);
+ when(patternConsumer.supportsTopicListWatcherReconcile()).thenReturn(true);
PatternConsumerUpdateQueue queue = new PatternConsumerUpdateQueue(patternConsumer, listener);
watcherFuture = new CompletableFuture<>();
watcher = new TopicListWatcher(queue, client,
TopicsPatternFactory.create(Pattern.compile(topic)), 7,
- NamespaceName.get("tenant/ns"), null, watcherFuture, () -> {});
+ NamespaceName.get("tenant/ns"), patternConsumer::getLocalStateTopicsHash, watcherFuture,
+ () -> 0);
}
@Test
@@ -94,34 +101,44 @@ public class TopicListWatcherTest {
@Test
public void testWatcherCreatesBrokerSideObjectWhenConnected() {
- ClientCnx clientCnx = mock(ClientCnx.class);
+ ClientCnx clientCnx = mockClientCnx();
CompletableFuture<CommandWatchTopicListSuccess> responseFuture = new CompletableFuture<>();
- ArgumentCaptor<BaseCommand> commandCaptor = ArgumentCaptor.forClass(BaseCommand.class);
- when(clientCnx.newWatchTopicList(any(BaseCommand.class), anyLong())).thenReturn(responseFuture);
+ when(clientCnx.newWatchTopicList(anyLong(), anyLong(), anyString(), anyString(), any()))
+ .thenReturn(responseFuture);
when(clientCnx.ctx()).thenReturn(mock(ChannelHandlerContext.class));
clientCnxFuture.complete(clientCnx);
- verify(clientCnx).newWatchTopicList(commandCaptor.capture(), anyLong());
+
+ verify(clientCnx).newWatchTopicList(anyLong(), anyLong(), anyString(), anyString(), any());
+
CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess()
.setWatcherId(7)
- .setRequestId(commandCaptor.getValue().getWatchTopicList().getRequestId())
+ .setRequestId(0)
.setTopicsHash("FEED");
success.addTopic("persistent://tenant/ns/topic11");
responseFuture.complete(success);
assertTrue(watcherFuture.isDone() && !watcherFuture.isCompletedExceptionally());
}
+ private static ClientCnx mockClientCnx() {
+ ClientCnx clientCnx = mock(ClientCnx.class);
+ when(clientCnx.isSupportsTopicWatchers()).thenReturn(true);
+ when(clientCnx.isSupportsTopicWatcherReconcile()).thenReturn(true);
+ return clientCnx;
+ }
+
@Test
public void testWatcherCallsListenerOnUpdate() {
- ClientCnx clientCnx = mock(ClientCnx.class);
+ ClientCnx clientCnx = mockClientCnx();
CompletableFuture<CommandWatchTopicListSuccess> responseFuture = new CompletableFuture<>();
- ArgumentCaptor<BaseCommand> commandCaptor = ArgumentCaptor.forClass(BaseCommand.class);
- when(clientCnx.newWatchTopicList(any(BaseCommand.class), anyLong())).thenReturn(responseFuture);
+ when(clientCnx.newWatchTopicList(anyLong(), anyLong(), anyString(), anyString(), any()))
+ .thenReturn(responseFuture);
when(clientCnx.ctx()).thenReturn(mock(ChannelHandlerContext.class));
clientCnxFuture.complete(clientCnx);
- verify(clientCnx).newWatchTopicList(commandCaptor.capture(), anyLong());
+ verify(clientCnx).newWatchTopicList(anyLong(), anyLong(), anyString(), anyString(), any());
+
CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess()
.setWatcherId(7)
- .setRequestId(commandCaptor.getValue().getWatchTopicList().getRequestId())
+ .setRequestId(0)
.setTopicsHash("FEED");
success.addTopic("persistent://tenant/ns/topic11");
responseFuture.complete(success);
@@ -134,4 +151,38 @@ public class TopicListWatcherTest {
watcher.handleCommandWatchTopicUpdate(update);
verify(listener).onTopicsAdded(Collections.singletonList("persistent://tenant/ns/topic12"));
}
+
+ @Test
+ public void testWatcherTriggersReconciliationOnHashMismatch() {
+ ClientCnx clientCnx = mockClientCnx();
+
+ CompletableFuture<CommandWatchTopicListSuccess> responseFuture = new CompletableFuture<>();
+ when(clientCnx.newWatchTopicList(anyLong(), anyLong(), anyString(), anyString(), any()))
+ .thenReturn(responseFuture);
+ when(clientCnx.ctx()).thenReturn(mock(ChannelHandlerContext.class));
+ clientCnxFuture.complete(clientCnx);
+
+ CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess()
+ .setWatcherId(7)
+ .setRequestId(0)
+ .setTopicsHash("FEED");
+ success.addTopic("persistent://tenant/ns/topic11");
+ responseFuture.complete(success);
+
+ // verify that the response was handled
+ verify(patternConsumer, times(1)).handleWatchTopicListSuccess(any(), any(), anyInt());
+ // sync local hash
+ when(patternConsumer.getLocalStateTopicsHash()).thenReturn("FEED");
+
+ // Send update with a mismatching hash
+ CommandWatchTopicUpdate update = new CommandWatchTopicUpdate()
+ .setTopicsHash("WRONG_HASH")
+ .setWatcherId(7)
+ .addAllNewTopics(Collections.singleton("persistent://tenant/ns/topic12"));
+
+ watcher.handleCommandWatchTopicUpdate(update);
+
+ // Verify that reconciliation was triggered again due to hash mismatch
+ verify(patternConsumer, times(1)).recheckTopicsChange();
+ }
}
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 deca71e9361..f32a35eb819 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
@@ -323,6 +323,7 @@ public class Commands {
connected.setFeatureFlags().setSupportsTopicWatchers(supportsTopicWatchers);
connected.setFeatureFlags().setSupportsGetPartitionedMetadataWithoutAutoCreation(true);
connected.setFeatureFlags().setSupportsReplDedupByLidAndEid(true);
+ connected.setFeatureFlags().setSupportsTopicWatcherReconcile(supportsTopicWatchers);
return cmd;
}
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 0e38f022472..3e10c1a1981 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
@@ -48,7 +48,7 @@ public class TopicList {
}
// get topics that match 'topicsPattern' from original topics list
- // return result should contain only topic names, without partition part
+ // matching is performed with base topic names, without partition part
public static List<String> filterTopics(List<String> original, String regex,
TopicsPattern.RegexImplementation topicsPatternImplementation) {
return filterTopics(original, TopicsPatternFactory.create(regex, topicsPatternImplementation));
diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto
index 1bfcf16beb5..edc382fc2d4 100644
--- a/pulsar-common/src/main/proto/PulsarApi.proto
+++ b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -315,6 +315,7 @@ message FeatureFlags {
optional bool supports_topic_watchers = 4 [default = false];
optional bool supports_get_partitioned_metadata_without_auto_creation = 5 [default = false];
optional bool supports_repl_dedup_by_lid_and_eid = 6 [default = false];
+ optional bool supports_topic_watcher_reconcile = 7 [default = false];
}
message CommandConnected {
@@ -653,7 +654,7 @@ message CommandTopicMigrated {
required ResourceType resource_type = 2;
optional string brokerServiceUrl = 3;
optional string brokerServiceUrlTls = 4;
-
+
}
@@ -1148,6 +1149,6 @@ message BaseCommand {
optional CommandWatchTopicListSuccess watchTopicListSuccess = 65;
optional CommandWatchTopicUpdate watchTopicUpdate = 66;
optional CommandWatchTopicListClose watchTopicListClose = 67;
-
+
optional CommandTopicMigrated topicMigrated = 68;
}
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java
index cec3392b385..f20216d57d8 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/client/impl/AutoCloseUselessClientConProxyTest.java
@@ -24,6 +24,7 @@ import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -41,6 +42,7 @@ import org.apache.pulsar.proxy.server.ProxyService;
import org.testng.annotations.Test;
@Test
+@Slf4j
public class AutoCloseUselessClientConProxyTest extends AutoCloseUselessClientConSupports {
private static final String TOPIC_NAME = BrokerTestUtil.newUniqueName("pattern_");
private static final String TOPIC_FULL_NAME = "persistent://public/default/" + TOPIC_NAME;
@@ -108,6 +110,9 @@ public class AutoCloseUselessClientConProxyTest extends AutoCloseUselessClientCo
.subscriptionName("my-subscription-y")
.subscribe();
+ waitForTopicListWatcherStarted(consumer);
+ waitForTopicListWatcherStarted(consumer2);
+
int poolSizeAfterCreatingConsumersAndProducer = pulsarClient.getCnxPool().getPoolSize();
// check that there are more than 3 connections
// at least 3 connections are required: