You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BackendArchitectX (via GitHub)" <gi...@apache.org> on 2026/10/17 20:21:20 UTC
[PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
BackendArchitectX opened a new pull request, #26969:
URL: https://github.com/apache/pulsar/pull/26969
Fixes #26802
### Motivation
`TopicResources.handleNotification` currently scans all registered topic listeners and evaluates their namespace patterns for each topic metadata notification. Notifications in one namespace therefore incur unnecessary matching work for watchers registered in unrelated namespaces.
This change makes topic notification dispatch scale primarily with the number of watchers in the affected namespace rather than the total number of registered watchers.
### Modifications
- Index topic listeners by namespace for direct lookup during persistent-topic creation and deletion notifications.
- Parse notification paths once instead of performing a regex match for every registered listener.
- Maintain listener registration and deregistration consistency under concurrent operations, without holding the registration lock during callbacks.
- Preserve session-event dispatch and backward compatibility with the deprecated `BiConsumer` listener APIs.
- Preserve filtering of system topics and reject irrelevant or malformed notification paths.
- Add regression and concurrency tests covering namespace isolation, listener lifecycle, re-registration, and notification behavior.
- Add a JMH benchmark comparing indexed dispatch with the original linear scan across different namespace and watcher counts.
### Verifying this change
- [x] `TopicResourcesTest`: 25 tests passed.
- [x] `TopicListServiceTest`: 16 tests passed.
- [x] Broker-common Checkstyle checks passed.
- [x] Microbenchmark Checkstyle and Java compilation passed.
- [x] Personal CI passed on https://github.com/BackendArchitectX/pulsar/pull/1 (43 successful checks, 1 skipped).
- [ ] Verify upstream Apache Pulsar CI after opening this PR.
The added JMH benchmark provides a reproducible way to compare dispatch behavior as the number and distribution of watchers increase.
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [x] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
**Threading model:** Listener registration and deregistration are coordinated to maintain consistent namespace indexing. Notification callbacks execute outside the registration lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#discussion_r4328701696
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java:
##########
@@ -1,3 +1,4 @@
+
Review Comment:
This empty line was introduced by the merge, and it breaks the Spotless license-header check:
```
> Task :pulsar-broker-common:spotlessJavaCheck FAILED
The following files had format violations:
src/main/java/org/apache/pulsar/broker/resources/TopicResources.java
@@ -1,3 +1,21 @@
+/*
+·*·Licensed·to·the·Apache·Software·Foundation·(ASF)·under·one
...
```
Please delete this line so the file starts with `/*` again, then confirm with `./gradlew :pulsar-broker-common:spotlessJavaCheck`. Note that `spotlessApply` won't fix this correctly: it would prepend a second license header above the blank line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #26969:
URL: https://github.com/apache/pulsar/pull/26969
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#issuecomment-6277514404
/pulsarbot rerun
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#issuecomment-6249374121
Thanks for the review and suggestions.
I've addressed the review feedback in commit `508a382`, added the subscription-notification benchmark and parser edge-case tests, and updated the PR description with the JMH results.
Both review conversations are resolved, all 27 `TopicResourcesTest` tests pass locally, and the branch has been updated with upstream `master`.
The required Pulsar CI workflows are currently awaiting maintainer approval. Could a maintainer please approve the workflow runs when convenient?
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on code in PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#discussion_r4306799981
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java:
##########
@@ -154,6 +167,317 @@ public void testNamespaceContainsDotsShouldntMatchAny() {
verify(listener).onTopicEvent("persistent://tenant/name.pace/topic", NotificationType.Created);
}
+ @Test
+ public void testListenersAreDispatchedOnlyWithinTheirNamespace() {
+ TopicListener first = mock(TopicListener.class);
+ TopicListener second = mock(TopicListener.class);
+ TopicListener other = mock(TopicListener.class);
+ when(first.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ when(second.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ when(other.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace2"));
+ topicResources.registerPersistentTopicListener(first);
+ topicResources.registerPersistentTopicListener(second);
+ topicResources.registerPersistentTopicListener(other);
+
+ topicResources.handleNotification(new Notification(NotificationType.Created,
+ "/managed-ledgers/tenant/namespace/persistent/topic"));
+
+ verify(first).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Created);
+ verify(second).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Created);
+ verify(other, never()).onTopicEvent(any(), any());
+ assertThat(topicResources.namespaceListenerBucketCount()).isEqualTo(2);
+ }
+
+ @Test
+ public void testInvalidMetadataPathsDoNotDispatchTopicEvents() {
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+
+ for (String path : List.of(
+ "/managed-ledgers/tenant/namespace/persistent/",
+ "/managed-ledgers/tenant/namespace/persistent/topic/subscription",
+ "/managed-ledgers/tenant/namespace/persistent/topic/",
+ "/managed-ledgers/tenant/namespace/non-persistent/topic",
+ "/managed-ledgers/tenant/namespace/segment/topic",
+ "/managed-ledgers/tenant/namespace/segment/topic/persistent/descriptor",
+ "/managed-ledgers-extra/tenant/namespace/persistent/topic",
+ "/managed-ledgers/tenant/namespace2/persistent/topic")) {
+ topicResources.handleNotification(new Notification(NotificationType.Created, path));
+ }
+
+ verify(listener).getNamespaceName();
+ verifyNoMoreInteractions(listener);
+ }
+
+ @Test
Review Comment:
Thanks for the suggestions. Addressed in commit `508a382`.
Added regression tests covering:
- A topic named `persistent`, including Created and Deleted notifications.
- Its subscription path, verifying that no topic callback is dispatched.
- A namespace and topic both named `persistent`.
- A legacy `tenant/cluster/ns` metadata path, verifying that it is ignored.
`TopicResourcesTest` now passes all 27 tests, and `checkstyleTest` passes.
I've also updated the PR description with the JMH results, including subscription-notification benchmark coverage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#issuecomment-6293587712
@BackendArchitectX Please resolve the merge conflicts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on code in PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#discussion_r4345129034
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java:
##########
@@ -1,3 +1,4 @@
+
Review Comment:
Thanks for catching this. I've removed the extra blank line in commit `5b6c5d98d4` and merged the latest upstream master.
Spotless, Checkstyle, `TopicResourcesTest`, `TopicListServiceTest`, and the microbenchmark build all pass locally.
The updated changes are pushed in commit `9657bc84f0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#discussion_r4305368286
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java:
##########
@@ -154,6 +167,317 @@ public void testNamespaceContainsDotsShouldntMatchAny() {
verify(listener).onTopicEvent("persistent://tenant/name.pace/topic", NotificationType.Created);
}
+ @Test
+ public void testListenersAreDispatchedOnlyWithinTheirNamespace() {
+ TopicListener first = mock(TopicListener.class);
+ TopicListener second = mock(TopicListener.class);
+ TopicListener other = mock(TopicListener.class);
+ when(first.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ when(second.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ when(other.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace2"));
+ topicResources.registerPersistentTopicListener(first);
+ topicResources.registerPersistentTopicListener(second);
+ topicResources.registerPersistentTopicListener(other);
+
+ topicResources.handleNotification(new Notification(NotificationType.Created,
+ "/managed-ledgers/tenant/namespace/persistent/topic"));
+
+ verify(first).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Created);
+ verify(second).onTopicEvent("persistent://tenant/namespace/topic", NotificationType.Created);
+ verify(other, never()).onTopicEvent(any(), any());
+ assertThat(topicResources.namespaceListenerBucketCount()).isEqualTo(2);
+ }
+
+ @Test
+ public void testInvalidMetadataPathsDoNotDispatchTopicEvents() {
+ TopicListener listener = mock(TopicListener.class);
+ when(listener.getNamespaceName()).thenReturn(NamespaceName.get("tenant/namespace"));
+ topicResources.registerPersistentTopicListener(listener);
+
+ for (String path : List.of(
+ "/managed-ledgers/tenant/namespace/persistent/",
+ "/managed-ledgers/tenant/namespace/persistent/topic/subscription",
+ "/managed-ledgers/tenant/namespace/persistent/topic/",
+ "/managed-ledgers/tenant/namespace/non-persistent/topic",
+ "/managed-ledgers/tenant/namespace/segment/topic",
+ "/managed-ledgers/tenant/namespace/segment/topic/persistent/descriptor",
+ "/managed-ledgers-extra/tenant/namespace/persistent/topic",
+ "/managed-ledgers/tenant/namespace2/persistent/topic")) {
+ topicResources.handleNotification(new Notification(NotificationType.Created, path));
+ }
+
+ verify(listener).getNamespaceName();
+ verifyNoMoreInteractions(listener);
+ }
+
+ @Test
Review Comment:
Optional: consider also covering a topic named `persistent` (`/managed-ledgers/tenant/namespace/persistent/persistent`) and its subscription path (`.../persistent/persistent/sub`), plus a legacy `tenant/cluster/ns` namespace. That would pin down how the `lastIndexOf("/persistent/")` parsing behaves.
##########
microbench/src/main/java/org/apache/pulsar/broker/resources/TopicResourcesNotificationBenchmark.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 static org.apache.pulsar.common.util.Codec.decode;
+import static org.mockito.Mockito.mock;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.naming.TopicName;
+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.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Compares indexed notification dispatch with the original per-listener namespace-pattern scan.
+ */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(1)
+public class TopicResourcesNotificationBenchmark {
+
+ @Param({"1", "100", "1000"})
+ public int namespaceCount;
+
+ @Param({"1", "10"})
+ public int watchersPerNamespace;
+
+ private TopicResources topicResources;
+ private Map<TopicListener, Pattern> originalListeners;
+ private Notification[] notifications;
+ private final AtomicInteger dispatchedEvents = new AtomicInteger();
+ private int notificationCursor;
+
+ @Setup
+ public void setup() {
+ topicResources = new TopicResources(mock(MetadataStoreExtended.class));
+ originalListeners = new ConcurrentHashMap<>();
+ notifications = new Notification[namespaceCount];
+
+ for (int namespaceIndex = 0; namespaceIndex < namespaceCount; namespaceIndex++) {
+ NamespaceName namespace = NamespaceName.get("tenant/namespace-" + namespaceIndex);
+ String namespacePath = namespace.toString();
+ notifications[namespaceIndex] = new Notification(NotificationType.Created,
Review Comment:
Could you add a case where the notification is a subscription node (`.../persistent/topic/sub`)? Before this PR, those paths ran every listener's regex and never matched, so that's where indexing should help most. Please also post the JMH results in the PR description (ideally on Linux x86_64, per `microbench/README.md`) so the benefit is quantified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Index topic watcher notifications by namespace [pulsar]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #26969:
URL: https://github.com/apache/pulsar/pull/26969#issuecomment-6296987033
@lhotari Thanks for pointing this out. I've merged the latest upstream `master` into the PR branch and resolved the conflicts. The PR is now mergeable.
Could you please take another look when you have a moment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org