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:54:56 UTC
(pulsar) branch branch-4.1 updated (c604310c88c -> eaf42b2569c)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from c604310c88c [improve][proxy] Add regression tests for package upload with 'Expect: 100-continue' (#25211)
new ab27ec90e8b [fix][test]Fix flaky ExtensibleLoadManagerImplTest_testGetMetrics (#25216)
new 2f474463abb [fix][broker] Fix ManagedCursorImpl.asyncDelete() method may lose previous async mark delete properties in race condition (#25165)
new 5bd426ac7b6 [fix][broker]Fix ledgerHandle failed to read by using new BK API (#25199)
new 2ad42019780 [fix][client] Fix producer synchronous retry handling in failPendingMessages method (#25207)
new eaf42b2569c [fix][broker] Prevent missed topic changes in topic watchers and schedule periodic refresh with patternAutoDiscoveryPeriod interval (#25188)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 7 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 4 +-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 109 ++++++
.../impl/ManagedLedgerFactoryShutdownTest.java | 6 +
.../pulsar/broker/resources/PulsarResources.java | 8 +-
.../pulsar/broker/resources/TopicListener.java | 10 +-
.../pulsar/broker/resources/TopicResources.java | 83 ++++-
.../broker/resources/TopicResourcesTest.java | 103 ++++--
.../bucket/BookkeeperBucketSnapshotStorage.java | 2 +-
.../pulsar/broker/service/TopicListService.java | 375 ++++++++++++++-------
.../service/schema/BookkeeperSchemaStorage.java | 2 +-
.../pulsar/compaction/CompactedTopicImpl.java | 2 +-
.../extensions/ExtensibleLoadManagerImplTest.java | 4 +-
.../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 ++++++-----
.../pulsar/client/impl/ProducerSyncRetryTest.java | 99 ++++++
.../org/apache/pulsar/client/impl/ClientCnx.java | 15 +
.../client/impl/PatternConsumerUpdateQueue.java | 176 +++++++---
.../impl/PatternMultiTopicsConsumerImpl.java | 216 ++++++++----
.../apache/pulsar/client/impl/ProducerImpl.java | 20 +-
.../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/ProducerImplTest.java | 77 +++++
.../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 +
37 files changed, 1961 insertions(+), 495 deletions(-)
copy pulsar-broker/src/main/java/org/apache/pulsar/broker/web/NoSwaggerDocumentation.java => pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicListener.java (69%)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSyncRetryTest.java
(pulsar) 05/05: [fix][broker] Prevent missed topic changes in topic watchers and schedule periodic refresh with patternAutoDiscoveryPeriod interval (#25188)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eaf42b2569c49a0129dc7fc194a45f1269bfb37a
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)
(cherry picked from commit 2e06cc03570d69f8a0df0271fd5178438ea06fff)
---
.../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 07df92f501e..35c49fb46a8 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());
@@ -1223,6 +1228,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 1d8ecb47278..85077c22c4c 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;
@@ -691,9 +692,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);
@@ -705,7 +711,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:
(pulsar) 02/05: [fix][broker] Fix ManagedCursorImpl.asyncDelete() method may lose previous async mark delete properties in race condition (#25165)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2f474463abb3a29978cd540ae1de41c75bb47af0
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Fri Feb 6 00:58:59 2026 +0800
[fix][broker] Fix ManagedCursorImpl.asyncDelete() method may lose previous async mark delete properties in race condition (#25165)
(cherry picked from commit bea6f8ac48740b44c5d162bd4aa0ca851eb91ae5)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 5 +-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 100 +++++++++++++++++++++
2 files changed, 101 insertions(+), 4 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index c120b9fa719..1ab6ceab9a2 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -2681,10 +2681,7 @@ public class ManagedCursorImpl implements ManagedCursor {
}
try {
- Map<String, Long> properties = lastMarkDeleteEntry != null ? lastMarkDeleteEntry.properties
- : Collections.emptyMap();
-
- internalAsyncMarkDelete(newMarkDeletePosition, properties, new MarkDeleteCallback() {
+ internalAsyncMarkDelete(newMarkDeletePosition, null, new MarkDeleteCallback() {
@Override
public void markDeleteComplete(Object ctx) {
callback.deleteComplete(ctx);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index bfb9b6ecca1..f41269b49af 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -5926,6 +5926,106 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(future1.get(2, TimeUnit.SECONDS).get(0).getData(), "msg".getBytes());
}
+ @Test(timeOut = 20000)
+ public void testAsyncMarkDeleteNeverLoseProperties() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(3);
+ config.setRetentionTime(20, TimeUnit.SECONDS);
+ config.setRetentionSizeInMB(5);
+
+ @Cleanup ManagedLedgerImpl ledger =
+ (ManagedLedgerImpl) factory.open("testAsyncMarkDeleteNeverLoseProperties", config);
+ @Cleanup ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1");
+
+ int numMessages = 20;
+ List<Position> positions = new ArrayList<>();
+ for (int i = 0; i < numMessages; i++) {
+ Position pos = ledger.addEntry("entry-1".getBytes(Encoding));
+ positions.add(pos);
+ }
+
+ String propertyKey = "test-property";
+ CountDownLatch latch = new CountDownLatch(numMessages);
+ for (int i = 0; i < numMessages; i++) {
+ Map<String, Long> properties = new HashMap<>();
+ properties.put(propertyKey, (long) i);
+ cursor.asyncMarkDelete(positions.get(i), properties, new MarkDeleteCallback() {
+ @Override
+ public void markDeleteComplete(Object ctx) {
+ latch.countDown();
+ }
+
+ @Override
+ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+ fail("Mark delete should succeed");
+ }
+ }, null);
+ }
+
+ latch.await();
+
+ int lastIndex = numMessages - 1;
+ assertEquals(cursor.getMarkDeletedPosition(), positions.get(lastIndex));
+ Map<String, Long> properties = cursor.getProperties();
+ assertEquals(properties.size(), 1);
+ assertEquals(properties.get(propertyKey), lastIndex);
+ }
+
+ @Test(timeOut = 20000)
+ public void testAsyncDeleteNeverLoseMarkDeleteProperties() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(11);
+
+ @Cleanup ManagedLedgerImpl ledger =
+ (ManagedLedgerImpl) factory.open("testAsyncDeleteNeverLoseMarkDeleteProperty", config);
+ @Cleanup ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1");
+
+ int numMessages = 10;
+ List<Position> positions = new ArrayList<>();
+ for (int i = 0; i < numMessages; i++) {
+ Position pos = ledger.addEntry("entry-1".getBytes(Encoding));
+ positions.add(pos);
+ }
+
+ String propertyKey = "test-property";
+ CountDownLatch latch = new CountDownLatch(numMessages);
+ for (int i = 0; i < numMessages - 1; i++) {
+ Map<String, Long> properties = new HashMap<>();
+ properties.put(propertyKey, (long) i);
+ cursor.asyncMarkDelete(positions.get(i), properties, new MarkDeleteCallback() {
+ @Override
+ public void markDeleteComplete(Object ctx) {
+ latch.countDown();
+ }
+
+ @Override
+ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+ fail("Mark delete should succeed");
+ }
+ }, null);
+ }
+
+ int lastIndex = numMessages - 1;
+ cursor.asyncDelete(positions.get(lastIndex), new DeleteCallback() {
+ @Override
+ public void deleteComplete(Object ctx) {
+ latch.countDown();
+ }
+
+ @Override
+ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
+ fail("Delete should succeed");
+ }
+ }, null);
+
+ latch.await();
+
+ assertEquals(cursor.getMarkDeletedPosition(), positions.get(lastIndex));
+ Map<String, Long> properties = cursor.getProperties();
+ assertEquals(properties.size(), 1);
+ assertEquals(properties.get(propertyKey), lastIndex - 1);
+ }
+
class TestPulsarMockBookKeeper extends PulsarMockBookKeeper {
Map<Long, Integer> ledgerErrors = new HashMap<>();
(pulsar) 01/05: [fix][test]Fix flaky ExtensibleLoadManagerImplTest_testGetMetrics (#25216)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ab27ec90e8b68e4c026837e62ee3aefa6c65a508
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Feb 6 00:53:25 2026 +0800
[fix][test]Fix flaky ExtensibleLoadManagerImplTest_testGetMetrics (#25216)
(cherry picked from commit 257d42ff9d4d7656a110d77013a89570c3feaa93)
---
.../broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index d7e85af1ca7..7962aaf3fa9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -1819,7 +1819,7 @@ public class ExtensibleLoadManagerImplTest extends ExtensibleLoadManagerImplBase
.get(3, TimeUnit.SECONDS);
}
- @Test
+ @Test(priority = Integer.MIN_VALUE)
public void testGetMetrics() throws Exception {
{
ServiceConfiguration conf = getDefaultConf();
@@ -2009,7 +2009,7 @@ public class ExtensibleLoadManagerImplTest extends ExtensibleLoadManagerImplBase
dimensions=[{broker=localhost, metric=sunitStateChn, result=Cancel}], metrics=[{brk_sunit_state_chn_inactive_broker_cleanup_ops_total=7}]
dimensions=[{broker=localhost, metric=sunitStateChn, result=Schedule}], metrics=[{brk_sunit_state_chn_inactive_broker_cleanup_ops_total=5}]
dimensions=[{broker=localhost, metric=sunitStateChn, result=Success}], metrics=[{brk_sunit_state_chn_inactive_broker_cleanup_ops_total=1}]
- dimensions=[{broker=localhost, metric=sunitStateChn}], metrics=[{brk_sunit_state_chn_orphan_su_cleanup_ops_total=3, brk_sunit_state_chn_owned_su_total=10, brk_sunit_state_chn_su_tombstone_cleanup_ops_total=2}]
+ dimensions=[{broker=localhost, metric=sunitStateChn}], metrics=[{brk_sunit_state_chn_last_metadata_session_event_age_seconds=-1, brk_sunit_state_chn_last_metadata_session_event_is_reestablished=1, brk_sunit_state_chn_last_metadata_session_event_timestamp_ms=0, brk_sunit_state_chn_metadata_state=0, brk_sunit_state_chn_orphan_su_cleanup_ops_total=3, brk_sunit_state_chn_owned_su_total=10, brk_sunit_state_chn_su_tombstone_cleanup_ops_total=2}]
dimensions=[{broker=localhost, metric=bundleUnloading}], metrics=[{brk_lb_ignored_ack_total=3, brk_lb_ignored_send_total=2}]
""".split("\n"));
var actual = primaryLoadManager.getMetrics().stream().map(Metrics::toString).collect(Collectors.toSet());
(pulsar) 04/05: [fix][client] Fix producer synchronous retry handling in failPendingMessages method (#25207)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2ad42019780c3a6f316718df71de86846eae44fe
Author: Malla Sandeep <sa...@gmail.com>
AuthorDate: Fri Feb 6 00:51:54 2026 +0530
[fix][client] Fix producer synchronous retry handling in failPendingMessages method (#25207)
(cherry picked from commit 611efe4a77a4cc9ec1875d4ee5fd29916d3f75c7)
---
.../pulsar/client/impl/ProducerSyncRetryTest.java | 99 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 20 +++--
.../pulsar/client/impl/ProducerImplTest.java | 77 +++++++++++++++++
3 files changed, 189 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSyncRetryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSyncRetryTest.java
new file mode 100644
index 00000000000..359db97fc78
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSyncRetryTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class ProducerSyncRetryTest extends ProducerConsumerBase {
+
+ @Override
+ @BeforeMethod
+ public void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @Override
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test(timeOut = 30000)
+ public void testProducerSyncRetryAfterTimeout() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/tp");
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableBatching(false)
+ .sendTimeout(1, TimeUnit.MILLISECONDS) // force timeout
+ .create();
+
+ // To make sure first message is timed out
+ this.stopBroker();
+
+ // First message will get timed out, then be retried with same payload
+ ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
+ MessageMetadata messageMetadata = new MessageMetadata();
+ messageMetadata.setUncompressedSize(1);
+ MessageImpl<byte[]> message = MessageImpl.create(messageMetadata, payload, Schema.BYTES, topic);
+
+ MessageMetadata retryMessageMetadata = new MessageMetadata();
+ retryMessageMetadata.setUncompressedSize(1);
+ MessageImpl<byte[]> retryMessage = MessageImpl.create(retryMessageMetadata, payload, Schema.BYTES, topic);
+
+ // First send is expected to fail
+ CompletableFuture<MessageId> firstSend = producer.sendAsync(message);
+ producer.triggerSendTimer();
+
+ // Waits until firstSend returns timeout exception
+ CompletableFuture<MessageId> retrySend =
+ firstSend.handle((msgId, ex) -> {
+ assertNotNull(ex, "First send must timeout");
+ assertTrue(ex instanceof PulsarClientException.TimeoutException);
+ try {
+ // Retry should succeed
+ this.startBroker();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ producer.conf.setSendTimeoutMs(10000);
+ return producer.sendAsync(retryMessage);
+ }).thenCompose(f -> f);
+
+ // Wait until retry completes successfully
+ MessageId retryMessageId = retrySend.join();
+ assertNotNull(retryMessageId);
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 589f7ef5ea7..89fa8ebb998 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -1820,11 +1820,12 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
messagesCount.set(0);
}
- public void remove() {
+ public OpSendMsg remove() {
OpSendMsg op = delegate.remove();
if (op != null) {
messagesCount.addAndGet(-op.numMessagesInBatch);
}
+ return op;
}
public OpSendMsg peek() {
@@ -2295,14 +2296,20 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
/**
- * This fails and clears the pending messages with the given exception. This method should be called from within the
- * ProducerImpl object mutex.
+ * This fails the pending messages at the start of the call, without dropping newly enqueued
+ * retry messages. This method should be called from within the ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
if (cnx == null) {
final AtomicInteger releaseCount = new AtomicInteger();
final boolean batchMessagingEnabled = isBatchMessagingEnabled();
- pendingMessages.forEach(op -> {
+ // Track message count to fail so that newly added messages by synchronous retries
+ // triggered by op.sendComplete(ex); don't get removed
+ int pendingMessagesToFailCount = pendingMessages.size();
+
+ for (int i = 0; i < pendingMessagesToFailCount; i++) {
+ OpSendMsg op = pendingMessages.remove();
releaseCount.addAndGet(batchMessagingEnabled ? op.numMessagesInBatch : 1);
try {
// Need to protect ourselves from any exception being thrown in the future handler from the
@@ -2322,9 +2329,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
- });
+ }
- pendingMessages.clear();
semaphoreRelease(releaseCount.get());
if (batchMessagingEnabled) {
failPendingBatchMessages(ex);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
index 40b76f37cc0..0fe177b775c 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
@@ -26,9 +26,12 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import java.nio.ByteBuffer;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.ByteBufPair;
import org.mockito.Mockito;
import org.testng.annotations.Test;
@@ -67,4 +70,78 @@ public class ProducerImplTest {
assertTrue(producer.populateMessageSchema(msg, null));
verify(msg).setSchemaState(MessageImpl.SchemaState.Ready);
}
+
+ @Test
+ public void testFailPendingMessagesSyncRetry()
+ throws Exception {
+ ProducerImpl<byte[]> producer =
+ Mockito.mock(ProducerImpl.class, Mockito.CALLS_REAL_METHODS);
+ // Disable batching
+ Mockito.doReturn(false)
+ .when(producer)
+ .isBatchMessagingEnabled();
+
+ // Stub semaphore release (not under test)
+ Mockito.doNothing()
+ .when(producer)
+ .semaphoreRelease(Mockito.anyInt());
+
+ // Stub client cleanup path (not under test)
+ PulsarClientImpl client = Mockito.mock(PulsarClientImpl.class);
+ Mockito.when(client.getMemoryLimitController())
+ .thenReturn(Mockito.mock(MemoryLimitController.class));
+ FieldUtils.writeField(producer, "client", client, true);
+
+ // Real pending queue
+ ProducerImpl.OpSendMsgQueue pendingQueue = new ProducerImpl.OpSendMsgQueue();
+ FieldUtils.writeField(producer, "pendingMessages", pendingQueue, true);
+
+ // OpSendMsg that retries reentrantly
+ MessageImpl<?> msg = Mockito.mock(MessageImpl.class);
+ Mockito.when(msg.getUncompressedSize()).thenReturn(10);
+ ProducerImpl.OpSendMsg op = ProducerImpl.OpSendMsg.create(
+ Mockito.mock(LatencyHistogram.class),
+ msg,
+ Mockito.mock(ByteBufPair.class),
+ 1L,
+ Mockito.mock(SendCallback.class)
+ );
+ op.totalChunks = 1;
+ op.chunkId = 0;
+ op.numMessagesInBatch = 1;
+
+ MessageImpl<?> retryMsg = Mockito.mock(MessageImpl.class);
+ Mockito.when(retryMsg.getUncompressedSize()).thenReturn(10);
+
+ // Override sendComplete to Reentrant retry via spy
+ ProducerImpl.OpSendMsg firstSpy = Mockito.spy(op);
+ Mockito.doAnswer(invocation -> {
+ // Reentrant retry during callback
+ ProducerImpl.OpSendMsg retryOp = ProducerImpl.OpSendMsg.create(
+ Mockito.mock(LatencyHistogram.class),
+ retryMsg,
+ Mockito.mock(ByteBufPair.class),
+ 2L,
+ Mockito.mock(SendCallback.class)
+ );
+ retryOp.totalChunks = 1;
+ retryOp.chunkId = 0;
+ retryOp.numMessagesInBatch = 1;
+ pendingQueue.add(retryOp);
+ return null;
+ }).when(firstSpy).sendComplete(Mockito.any());
+ Mockito.doNothing()
+ .when(firstSpy)
+ .recycle();
+
+ // Seed initial pending message
+ pendingQueue.add(firstSpy);
+
+ // Invoke failPendingMessages(null, ex)
+ producer.failPendingMessages(null, new PulsarClientException.TimeoutException("timeout"));
+ assertEquals(producer.getPendingQueueSize(), 1,
+ "Retry Op should exist in the pending Queue");
+ assertEquals(pendingQueue.peek().sequenceId, 2L,
+ "Retry Op SequenceId should match with the one in pendingQueue");
+ }
}
(pulsar) 03/05: [fix][broker]Fix ledgerHandle failed to read by using new BK API (#25199)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5bd426ac7b6bdb57ffec498fb95c1ab0842f4129
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Feb 6 01:01:16 2026 +0800
[fix][broker]Fix ledgerHandle failed to read by using new BK API (#25199)
(cherry picked from commit 6d51f8883206fd81840faddf4cec840f6969b137)
---
.../org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 2 +-
.../org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 4 ++--
.../org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java | 9 +++++++++
.../mledger/impl/ManagedLedgerFactoryShutdownTest.java | 6 ++++++
.../broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java | 2 +-
.../pulsar/broker/service/schema/BookkeeperSchemaStorage.java | 2 +-
.../java/org/apache/pulsar/compaction/CompactedTopicImpl.java | 2 +-
7 files changed, 21 insertions(+), 6 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 1ab6ceab9a2..7009ac750fe 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -686,7 +686,7 @@ public class ManagedCursorImpl implements ManagedCursor {
};
try {
bookkeeper.asyncOpenLedger(ledgerId, digestType, getConfig().getPassword(), openCallback,
- null);
+ null, true);
} catch (Throwable t) {
log.error("[{}] Encountered error on opening cursor ledger {} for cursor {}",
ledger.getName(), ledgerId, name, t);
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 06a3a5a4e56..27ea422ef6a 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -481,7 +481,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.debug("[{}] Opening ledger {}", name, id);
}
mbean.startDataLedgerOpenOp();
- bookKeeper.asyncOpenLedger(id, digestType, config.getPassword(), opencb, null);
+ bookKeeper.asyncOpenLedger(id, digestType, config.getPassword(), opencb, null, true);
} else {
initializeBookKeeper(callback);
}
@@ -1908,7 +1908,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
handleBadVersion(new BadVersionException("the current ledger " + currentLedger.getId()
+ " was concurrent modified by a other bookie client. The error code is: " + errorCode));
}
- }, null);
+ }, null, true);
}
synchronized void ledgerClosed(final LedgerHandle lh) {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index f41269b49af..522e1290320 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -6045,6 +6045,15 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
super.asyncOpenLedger(lId, digestType, passwd, cb, ctx);
}
}
+
+ public void asyncOpenLedger(final long lId, final DigestType digestType, final byte[] passwd,
+ final OpenCallback cb, final Object ctx, boolean keepMetadataUpdate) {
+ if (ledgerErrors.containsKey(lId)) {
+ cb.openComplete(ledgerErrors.get(lId), null, ctx);
+ } else {
+ super.asyncOpenLedger(lId, digestType, passwd, cb, ctx, keepMetadataUpdate);
+ }
+ }
}
private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryShutdownTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryShutdownTest.java
index ecc3423e292..95f0a6b8c77 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryShutdownTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryShutdownTest.java
@@ -19,6 +19,7 @@
package org.apache.bookkeeper.mledger.impl;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
@@ -137,6 +138,11 @@ public class ManagedLedgerFactoryShutdownTest {
cb.openComplete(0, ledgerHandle, inv.getArgument(4, Object.class));
return null;
}).when(bookKeeper).asyncOpenLedger(anyLong(), any(), any(), any(), any());
+ doAnswer(inv -> {
+ AsyncCallback.OpenCallback cb = inv.getArgument(3, AsyncCallback.OpenCallback.class);
+ cb.openComplete(0, ledgerHandle, inv.getArgument(4, Object.class));
+ return null;
+ }).when(bookKeeper).asyncOpenLedger(anyLong(), any(), any(), any(), any(), anyBoolean());
doAnswer(inv -> {
AsyncCallback.CreateCallback cb = inv.getArgument(5, AsyncCallback.CreateCallback.class);
cb.createComplete(0, newLedgerHandle, inv.getArgument(6, Object.class));
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
index ba37092e88d..fa7408d7e15 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
@@ -213,7 +213,7 @@ public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
} else {
future.complete(handle);
}
- }, null
+ }, null, true
);
return future;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
index b931239a32c..e38bf48f1fd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -630,7 +630,7 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
} else {
future.complete(handle);
}
- }, null
+ }, null, true
);
return future;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index 21cc0e45acd..2ceaab83f22 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -184,7 +184,7 @@ public class CompactedTopicImpl implements CompactedTopic {
} else {
promise.complete(ledger);
}
- }, null);
+ }, null, true);
return promise.thenApply((ledger) -> new CompactedTopicContext(
ledger, createCache(ledger, DEFAULT_MAX_CACHE_SIZE)));
}