[feat][broker] PIP-145: Notifications for faster topic discovery #16062

andrasbeni · 2022-06-14T16:37:34Z

Motivation

When using a pattern-based subscription, clients poll for new topics that
match the pattern. It can happen frequently that the first messages produced
to the topic will be consumed with a significant delay because the client
has not got information about the topic for a while.

This commit introduces topic list watchers. By using these objects
clients act as observers of the creation or deletion of topics closer to
real-time. This reduces latency in consuming the first messages
published to a topic when using a pattern-based subscription.

Modifications

New commands were added to the binary protocol to enable registering
and deregistering watchers.
Pattern-based consumers create TopicListWatcher objects if the broker
supports this feature. Otherwise, they fall back to polling only.
The watchers use ConnectionHandler to obtain a connection to a broker.
Once connected, watchers register and wait for updates.
ServerCnx uses the newly created TopicListService to manage watchers.
TopicListService listens to metadata notifications and sends updates.

Verifying this change

Make sure that the change passes the CI checks.

This change added tests

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API: no
The schema: no
The default values of configurations: no
The wire protocol: yes
- Adds new commands and a new optional field to an existing command
The rest endpoints: no
The admin cli options: no
Anything that affects deployment: no

Documentation

Check the box below or label this PR directly.

Need to update docs?

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

This commit introduces topic list watchers. By using these objects clients can observe the creation or deletion of topics closer to real-time. This reduces latency in consuming the first messages published to a topic when using a pattern-based subscription. Modifications: - New commands were added to the binary protocol to enable registering and deregistering watchers. - Pattern-based consumers create TopicListWatcher objects if the broker supports this feature. Otherwise, they fall back to polling only. - The watchers use ConnectionHandler to obtain a connection to a broker. - Once connected, watchers register and wait for updates. - ServerCnx uses the newly created TopicListService to manage watchers. - TopicListService listens to metadata notifications and sends updates.

codelipenghui

Overall looks pretty good. Just left some minor comments.

codelipenghui · 2022-06-15T11:49:20Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

    // These sets are updated from multiple threads, so they require a threadsafe data structure
    private final Set<ProducerBase<?>> producers = Collections.newSetFromMap(new ConcurrentHashMap<>());
    private final Set<ConsumerBase<?>> consumers = Collections.newSetFromMap(new ConcurrentHashMap<>());
+    private final Set<TopicListWatcher> topicListWatchers = Collections.newSetFromMap(new ConcurrentHashMap<>());


Looks like no one add watchers to this Set?

You are right. This set is redundant to ClientCnx.topicListWatchers, so I'm going to remove this.

codelipenghui · 2022-06-15T12:01:31Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java

+                            watcherFuture.completeExceptionally(
+                                    PulsarClientException.wrap(e, String.format("Failed to create topic list watcher %s"
+                                                    + "when connecting to the broker", getHandlerName())));
+                            client.cleanupTopicListWatcher(this);


Should also cleanup the watcher from the ClientCnx.topicListWatchers since line:123 added the watcher to the ClientCnx.topicListWatchers

deregisterFromClientCnx(), this exceptionally block's first call does call ClientCnx.removeTopicListWatcher(long watcherId), so I think it should be fine.

codelipenghui · 2022-06-15T12:05:56Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

+    public CompletableFuture<CommandWatchTopicListSuccess> newUnwatchTopicList(
+            BaseCommand commandUnwatchTopicList, long requestId) {
+        if (!supportsTopicWatchers) {
+            return FutureUtil.failedFuture(
+                    new PulsarClientException.NotAllowedException(
+                            "Broker does not allow broker side pattern evaluation."));
+        }
+        return sendRequestAndHandleTimeout(Commands.serializeWithSize(commandUnwatchTopicList), requestId,
+                RequestType.Command, true);
+    }


Looks this method is unused. The TopicListWatcher uses Commands.newUnwatchTopicList directly.

Nice catch. I am removing it.

codelipenghui · 2022-06-15T12:09:56Z

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

+        BaseCommand cmd = localCmd(Type.WATCH_TOPIC_LIST);
+        cmd.setWatchTopicList()
+                .setRequestId(requestId)
+                .setNamespace(namespace)


Can we make the namespace optional? Since we might support pattern subscribe without namespace/tenant. A related PR #7855

Or maybe better to have tenant and namespace, just keep them optional.

Yes, that feature would make sense to me. However, the linked pull request has seen no progress for almost two years, so if you don't mind I would like to keep the namespace parameter, for now, to simplify detecting when a metadata store notification is related to topics that match. Otherwise, it is more difficult to distinguish between topics in v1 namespaces and subscriptions in v2 namespaces. Or would you have a simple solution to that?

codelipenghui · 2022-06-15T12:38:31Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java

+            // create operation will complete, the new watcher will be discarded.
+            log.info("[{}] Closed watcher before its creation was completed. watcherId={}",
+                    connection.getRemoteAddress(), watcherId);
+            return;


Any reason that doesn't remove from the watchers map if reach here?

codelipenghui · 2022-06-15T12:38:53Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java

+        if (watcherFuture.isCompletedExceptionally()) {
+            log.info("[{}] Closed watcher that already failed to be created. watcherId={}",
+                    connection.getRemoteAddress(), watcherId);
+            return;


Same as the above comment.

codelipenghui · 2022-06-15T12:43:02Z

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java

+            CompletableFuture<CommandWatchTopicListSuccess> result =
+                    cnx.newWatchTopicList(Commands.newWatchTopicList(7, 5, "tenant/ns",
+                            ".*", null), 7);
+            assertTrue(result.isCompletedExceptionally());


Please also check the watcher is cleaned up if server does not support.

merlimat

Looks good! Left just few comments

merlimat · 2022-06-23T17:52:29Z

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java

+        }
+    }
+
+    Pattern namespaceNameToTopicNamePattern(NamespaceName namespaceName) {


We could have this as part of NamespaceName where we can store the compiled pattern.

That would improve performance indeed. The only problem I'd have with it is that we'd be referring to a relatively low-level detail (the managed ledger path) in a very general and high-level class. To me, it feels like breaking encapsulation/abstraction.
What do you think about caching the patterns in this class instead? Something along the lines of NamespaceName.cache.

In TopicName we already have getPersistenceName() which is kind of similar. On one hand I agree that it's not directly related to the logical name, though it was a way to ensure all the naming related ops are in a single place with no duplicated approaches.

merlimat · 2022-06-23T18:41:28Z

pulsar-common/src/main/proto/PulsarApi.proto

+        WATCH_TOPIC_LIST = 64;
+        WATCH_TOPIC_LIST_SUCCESS = 65;
+        WATCH_TOPIC_UPDATE = 66;
+        UNWATCH_TOPIC_LIST = 67;


nit: maybe we could rename to WATCH_TOPIC_CLOSE to have the same prefix

merlimat · 2022-06-23T18:50:44Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

+            }
+            isNamespaceOperationAllowed(namespaceName, NamespaceOperation.GET_TOPICS).thenApply(isAuthorized -> {
+                if (isAuthorized) {
+                    topicListService.handleWatchTopicList(commandWatchTopicList, lookupSemaphore);


This method will potentially get called from a different thread (if the authz is not already cached), therefore we cannot pass on the commandWatchTopicList object because it will get reused when the io-thread is processing the next call.

We need to extract the fields that we are interested in into local variables.

Thanks for pointing this out. I haven't thought of it.

merlimat · 2022-06-23T18:51:34Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

+    }
+
+    protected void handleCommandUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) {
+        topicListService.handleUnwatchTopicList(commandUnwatchTopicList);


Should we check authorization here or it will be not necessary?

We don't need to check authorization because executing this command only alters the state of TopicListService, which is in a 1:1 relation with ServerCnx. So it belongs to a single client, which was already checked when it created the watcher. (And if it did not have the authz to create one, it will be unable to delete anything)
Similar to handleCloseConsumer.

…nt modification

merlimat

👍

…che#16062) * PIP-145: Notifications for faster topic discovery This commit introduces topic list watchers. By using these objects clients can observe the creation or deletion of topics closer to real-time. This reduces latency in consuming the first messages published to a topic when using a pattern-based subscription. Modifications: - New commands were added to the binary protocol to enable registering and deregistering watchers. - Pattern-based consumers create TopicListWatcher objects if the broker supports this feature. Otherwise, they fall back to polling only. - The watchers use ConnectionHandler to obtain a connection to a broker. - Once connected, watchers register and wait for updates. - ServerCnx uses the newly created TopicListService to manage watchers. - TopicListService listens to metadata notifications and sends updates. * Fix checkstyle violation * Fix cpp client compile error * Remove unused code, remove failed watchers * Rename command Unwatch, extract fields from command to avoid concurrent modification * Fix cpp client compile error * Fix cpp client compile error Co-authored-by: Matteo Merli <mmerli@apache.org>

…ery (apache#16062)" This reverts commit d9f5640.

…che#16062) * PIP-145: Notifications for faster topic discovery This commit introduces topic list watchers. By using these objects clients can observe the creation or deletion of topics closer to real-time. This reduces latency in consuming the first messages published to a topic when using a pattern-based subscription. Modifications: - New commands were added to the binary protocol to enable registering and deregistering watchers. - Pattern-based consumers create TopicListWatcher objects if the broker supports this feature. Otherwise, they fall back to polling only. - The watchers use ConnectionHandler to obtain a connection to a broker. - Once connected, watchers register and wait for updates. - ServerCnx uses the newly created TopicListService to manage watchers. - TopicListService listens to metadata notifications and sends updates. * Fix checkstyle violation * Fix cpp client compile error * Remove unused code, remove failed watchers * Rename command Unwatch, extract fields from command to avoid concurrent modification * Fix cpp client compile error * Fix cpp client compile error Co-authored-by: Matteo Merli <mmerli@apache.org>

merlimat added this to the 2.11.0 milestone Jun 14, 2022

Anonymitaet added the doc-required Your PR changes impact docs and you will update later. label Jun 15, 2022

Andras Beni added 2 commits June 15, 2022 08:24

Fix checkstyle violation

1f4d9ad

Fix cpp client compile error

577c717

codelipenghui requested review from Technoboy-, congbobo184, eolivelli, hangc0276, mattisonchao, merlimat and michaeljmarshall June 15, 2022 09:03

codelipenghui reviewed Jun 15, 2022

View reviewed changes

codelipenghui added type/feature The PR added a new feature or issue requested a new feature area/broker component/client-java labels Jun 15, 2022

codelipenghui assigned andrasbeni Jun 15, 2022

Remove unused code, remove failed watchers

3a97b69

merlimat reviewed Jun 23, 2022

View reviewed changes

Andras Beni added 3 commits June 24, 2022 17:32

Rename command Unwatch, extract fields from command to avoid concurre…

a177ade

…nt modification

Fix cpp client compile error

da7d5c4

Fix cpp client compile error

c71bb64

merlimat approved these changes Jun 27, 2022

View reviewed changes

codelipenghui approved these changes Jun 28, 2022

View reviewed changes

Merge remote-tracking branch 'apache/master' into PIP-145-Part2

78471b6

merlimat added doc-not-needed Your PR changes do not impact docs and removed doc-required Your PR changes impact docs and you will update later. labels Jul 1, 2022

merlimat merged commit d9f5640 into apache:master Jul 1, 2022

codelipenghui added a commit to codelipenghui/incubator-pulsar that referenced this pull request Jul 12, 2022

Revert "[feat][broker] PIP-145: Notifications for faster topic discov…

ec9bb8f

…ery (apache#16062)" This reverts commit d9f5640.

andrasbeni mentioned this pull request Jul 13, 2022

[fix][java-client] Skip TopicListWatcher creation when lookup url scheme is HTTP #16576

Merged

2 tasks

shibd mentioned this pull request Jul 21, 2022

[fix][client] Fix subscription topic name error. #16719

Merged

1 task

andrasbeni deleted the PIP-145-Part2 branch August 8, 2022 13:09

shibd mentioned this pull request Mar 12, 2025

[fix][broker] Pattern subscription doesn't work when the pattern excludes the topic domain. #24072

Merged

4 tasks

[feat][broker] PIP-145: Notifications for faster topic discovery #16062

[feat][broker] PIP-145: Notifications for faster topic discovery #16062

Uh oh!

Conversation

andrasbeni commented Jun 14, 2022

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

codelipenghui left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

merlimat left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

merlimat left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants