You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "coderzc (via GitHub)" <gi...@apache.org> on 2026/01/11 09:46:16 UTC
[PR] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
coderzc opened a new pull request, #25134:
URL: https://github.com/apache/pulsar/pull/25134
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
<!-- Either this PR fixes an issue, -->
Fixes #xyz
<!-- or this PR is one task of an issue -->
Main Issue: #xyz
<!-- If the PR belongs to a PIP, please add the PIP link here -->
PIP: #xyz
<!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/pip/README.md -->
### Motivation
<!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
### Modifications
<!-- Describe the modifications you've done. -->
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [x] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [x] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2691233659
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.topics();
+ boolean isAlreadyFiltered = listingResult.filtered();
+
+ // Apply regex filtering (if the interceptor did not filter and regex matching is enabled)
+ if (!isAlreadyFiltered && enableSubscriptionPatternEvaluation
+ && topicsPattern.isPresent()) {
+ ......
+ }
+
+ // Calculate the Hash and send the response
+ String hash = TopicList.calculateHash(finalTopics);
+ boolean hashUnchanged = topicsHash.isPresent() && topicsHash.get().equals(hash);
+ ......
+ },
+ // Handle logic for heap memory limit exceeded inside withUpdatedPermits
+ t -> {
+ ......
+ return CompletableFuture.completedFuture(null);
+ });
+ }).whenComplete((__, ___) -> {
+ lookupSemaphore.release();
+ }).exceptionally(ex -> {
+ ......
+ return null;
+ });
+ },
+ // Handle logic for initial acquire failure
+ t -> {
+ ......
+ });
+ });
+}
+```
+
+### Client Changes
+Update LookupService
+The internal LookupService interface is updated to accept the properties map.
+
+```Java
+CompletableFuture<List<String>> getTopicsUnderNamespace(
+ NamespaceName namespace,
+ Mode mode,
+ String topicsPattern,
+ String topicsHash,
+ Map<String, String> properties
+);
+```
+Update `PatternMultiTopicsConsumerImpl`
+The regex consumer implementation (PatternMultiTopicsConsumerImpl) will be updated to extract the properties from ConsumerConfigurationData and pass them to the LookupService.
+
+```Java
+// In PatternMultiTopicsConsumerImpl.java
+ConsumerConfigurationData conf;
+Map<String, String> contextProperties = conf.getProperties();
+
+lookup.getTopicsUnderNamespace(
+ namespace,
+ mode,
+ topicsPattern.pattern(),
+ topicsHash,
+ contextProperties // Pass consumer properties here
+).thenAccept(topics -> {
+ // ... update subscriptions ...
+});
+```
+
+
+# Backward Compatibility
+Protocol: Adding an optional field (properties) to the Protobuf definition is non-breaking. Old clients will not send this field; old brokers will ignore it.
+
+Behavior: The default strategy mimics the current behavior exactly. The Broker retains the final filtering logic, ensuring that even custom strategies cannot return topics that violate the client's requested pattern.
+
+# Security Considerations
+Input Validation: The properties map is user-supplied input. Implementers of the customizable strategy MUST validate and sanitize these inputs before using them, especially if they are used to construct database queries.
+
+Authorization: This PIP only controls the discovery (listing) of topics. It does not bypass the Authorization Service for consuming or producing to those topics.
+
Review Comment:
Answering my own question about the topic properties. They were added with [PIP-110](https://github.com/apache/pulsar/blob/master/pip/pip-110.md), #12818 .
The topic properties for partitioned topics are stored in PartitionedTopicMetadata,
https://github.com/apache/pulsar/blob/a373258d0430aa04cc05c4c46b0ac5e0e38153cb/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/partition/PartitionedTopicMetadata.java#L33-L34
However, they are stored in LedgerMetadata for non-partitioned topics:
https://github.com/apache/pulsar/blob/75658cc353e74e132ef30090767fe062eae9ca05/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java#L665-L666
(PR #17238)
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689745860
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.topics();
+ boolean isAlreadyFiltered = listingResult.filtered();
+
+ // Apply regex filtering (if the interceptor did not filter and regex matching is enabled)
+ if (!isAlreadyFiltered && enableSubscriptionPatternEvaluation
+ && topicsPattern.isPresent()) {
+ ......
+ }
+
+ // Calculate the Hash and send the response
+ String hash = TopicList.calculateHash(finalTopics);
+ boolean hashUnchanged = topicsHash.isPresent() && topicsHash.get().equals(hash);
+ ......
+ },
+ // Handle logic for heap memory limit exceeded inside withUpdatedPermits
+ t -> {
+ ......
+ return CompletableFuture.completedFuture(null);
+ });
+ }).whenComplete((__, ___) -> {
+ lookupSemaphore.release();
+ }).exceptionally(ex -> {
+ ......
+ return null;
+ });
+ },
+ // Handle logic for initial acquire failure
+ t -> {
+ ......
+ });
+ });
+}
+```
+
+### Client Changes
+Update LookupService
+The internal LookupService interface is updated to accept the properties map.
+
+```Java
+CompletableFuture<List<String>> getTopicsUnderNamespace(
+ NamespaceName namespace,
+ Mode mode,
+ String topicsPattern,
+ String topicsHash,
+ Map<String, String> properties
+);
+```
+Update `PatternMultiTopicsConsumerImpl`
+The regex consumer implementation (PatternMultiTopicsConsumerImpl) will be updated to extract the properties from ConsumerConfigurationData and pass them to the LookupService.
+
+```Java
+// In PatternMultiTopicsConsumerImpl.java
+ConsumerConfigurationData conf;
+Map<String, String> contextProperties = conf.getProperties();
+
+lookup.getTopicsUnderNamespace(
+ namespace,
+ mode,
+ topicsPattern.pattern(),
+ topicsHash,
+ contextProperties // Pass consumer properties here
+).thenAccept(topics -> {
+ // ... update subscriptions ...
+});
+```
+
+
+# Backward Compatibility
+Protocol: Adding an optional field (properties) to the Protobuf definition is non-breaking. Old clients will not send this field; old brokers will ignore it.
+
+Behavior: The default strategy mimics the current behavior exactly. The Broker retains the final filtering logic, ensuring that even custom strategies cannot return topics that violate the client's requested pattern.
+
+# Security Considerations
+Input Validation: The properties map is user-supplied input. Implementers of the customizable strategy MUST validate and sanitize these inputs before using them, especially if they are used to construct database queries.
+
+Authorization: This PIP only controls the discovery (listing) of topics. It does not bypass the Authorization Service for consuming or producing to those topics.
+
Review Comment:
"Alternative solutions" section is missing.
I'm wondering if it would have made sense to extend the MetadataStore interfaces to support properties.
The problem that this PIP-452 causes is that it will be a burden to add additional custom interfaces to add properties to topics and so on. If the MetadataStore interface would support properties, it would be a cleaner approach to close the gaps.
Has the solution for adding properties to MetadataStore been considered? What are the reasons for rejecting this alternative?
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2755421178
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
Review Comment:
Ok, so can we remove all the related information for `enableBrokerTopicListWatcher`? I mean, after we supported `topic list watching for customized topic listing`, this configuration will be useless?
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
Review Comment:
If it's required to be added, please help provide more information about how this configuration works. I'm not pretty clear about this part.
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
Review Comment:
Just checked the Kubernetes style
```
kubectl get pods -l app=nginx,environment=production
```
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684574485
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
Since the built-in implementation is very simple that calls method directly on the `NamespaceService`, I think the fallback logic is unnecessary.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#issuecomment-3823439467
After discussions with @codelipenghui and @BewareMyPower, we found that the current working method of `BrokerInterceptor` cannot accurately specify which result to use when multiple BrokerInterceptors are configured. Therefore, a new pluggable interface `PulsarResourcesExtended` has been introduced to implement this feature.
`PulsarResourcesExtended` is defined as a general interface used as an extension for custom PulsarResources
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc merged PR #25134:
URL: https://github.com/apache/pulsar/pull/25134
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734668783
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
Review Comment:
done
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684569996
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
This design is a bit over-abstraction. IMO, the following design could be simpler and clearer.
```java
public record TopicListingResult(List<String> topics, boolean filtered) {}
```
```java
/**
* @return the future of the result, if it's empty, fall back to the built-in implementation to list all topics
*/
default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
```
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684561910
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
+ }
+
+ /**
+ * Return a successful result with the provided topics list.
+ */
+ public static TopicListingResult success(List<String> topics, boolean filtered) {
+ if (topics == null) {
+ throw new IllegalArgumentException("Topics list cannot be null in success result");
+ }
+ return new TopicListingResult(topics, filtered);
+ }
+
+ /**
+ * Check if this result is a "Pass Through" result.
+ */
+ public boolean isPassThrough() {
+ return this == PASS_THROUGH_INSTANCE;
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * <ul>
+ * <li>{@link TopicListingResult#passThrough()} :
+ * (Default) Proceed to the next interceptor or Broker's default logic.</li>
+ * <li>{@link TopicListingResult#success(java.util.List, boolean)} :
+ * Stop the chain and return these topics immediately.</li>
+ * </ul>
+ * If the future completes with {@code null}, it is treated as {@code passThrough()}.
+ */
+ default CompletableFuture<TopicListingResult> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(TopicListingResult.passThrough());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor
+ CompletableFuture<TopicListingResult> interceptorFuture =
+ getBrokerService().getPulsar().getBrokerInterceptor()
+ .interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && !interceptorResult.isPassThrough()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> TopicListingResult.success(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ List<String> rawTopics = listingResult.getTopics();
Review Comment:
You don't have to copy so much code here, which does not focus on the key change. The only thing that matters is that what will replace the following existing code:
```java
return getBrokerService().pulsar().getNamespaceService()
.getListOfUserTopics(namespaceName, mode)
```
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689703446
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
Review Comment:
Please mention this in the PIP-452 document since this is relevant context.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734536553
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
Review Comment:
Yes, this query behavior is determined by the plug-in, and it is not certain how many keys are contained in the properties.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734668517
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
Review Comment:
done
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
Review Comment:
done
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684569996
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
This design is unnecessary over-engineering. IMO, the following design could be simpler and clearer.
```java
public record TopicListingResult(List<String> topics, boolean filtered) {}
```
```java
/**
* @return the future of the result, if it's empty, fall back to the built-in implementation to list all topics
*/
default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
```
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2680899806
##########
pip/pip-452.md:
##########
@@ -0,0 +1,212 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Introduce a `NamespaceTopicListingStrategy` SPI to customize how topics are retrieved.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will abstract the topic retrieval logic into a NamespaceTopicListingStrategy.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+1. Protocol Changes
+We update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+2. Broker Changes
+New Interface: `NamespaceTopicListingStrategy`
+We introduce a strategy interface in the broker.
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+
+public class NamespaceTopicListingResult {
+ private final List<String> topics;
+
+ // Indicates if the plugin has already applied the regex filtering provided in the request.
+ // If true, the Broker will SKIP the subsequent filtering step.
+ private final boolean filtered;
+
+ public NamespaceTopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+ public List<String> getTopics() {
+ return topics;
+ }
+
+ public boolean isFiltered() {
+ return filtered;
+ }
+}
+```
+
+```Java
+
+package org.apache.pulsar.broker.namespace;
+
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface NamespaceTopicListingStrategy {
Review Comment:
It's better to extend the existing interceptor rather than introducing yet another plugin
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689910139
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.topics();
+ boolean isAlreadyFiltered = listingResult.filtered();
+
+ // Apply regex filtering (if the interceptor did not filter and regex matching is enabled)
+ if (!isAlreadyFiltered && enableSubscriptionPatternEvaluation
+ && topicsPattern.isPresent()) {
+ ......
+ }
+
+ // Calculate the Hash and send the response
+ String hash = TopicList.calculateHash(finalTopics);
+ boolean hashUnchanged = topicsHash.isPresent() && topicsHash.get().equals(hash);
+ ......
+ },
+ // Handle logic for heap memory limit exceeded inside withUpdatedPermits
+ t -> {
+ ......
+ return CompletableFuture.completedFuture(null);
+ });
+ }).whenComplete((__, ___) -> {
+ lookupSemaphore.release();
+ }).exceptionally(ex -> {
+ ......
+ return null;
+ });
+ },
+ // Handle logic for initial acquire failure
+ t -> {
+ ......
+ });
+ });
+}
+```
+
+### Client Changes
+Update LookupService
+The internal LookupService interface is updated to accept the properties map.
+
+```Java
+CompletableFuture<List<String>> getTopicsUnderNamespace(
+ NamespaceName namespace,
+ Mode mode,
+ String topicsPattern,
+ String topicsHash,
+ Map<String, String> properties
+);
+```
+Update `PatternMultiTopicsConsumerImpl`
+The regex consumer implementation (PatternMultiTopicsConsumerImpl) will be updated to extract the properties from ConsumerConfigurationData and pass them to the LookupService.
+
+```Java
+// In PatternMultiTopicsConsumerImpl.java
+ConsumerConfigurationData conf;
+Map<String, String> contextProperties = conf.getProperties();
+
+lookup.getTopicsUnderNamespace(
+ namespace,
+ mode,
+ topicsPattern.pattern(),
+ topicsHash,
+ contextProperties // Pass consumer properties here
+).thenAccept(topics -> {
+ // ... update subscriptions ...
+});
+```
+
+
+# Backward Compatibility
+Protocol: Adding an optional field (properties) to the Protobuf definition is non-breaking. Old clients will not send this field; old brokers will ignore it.
+
+Behavior: The default strategy mimics the current behavior exactly. The Broker retains the final filtering logic, ensuring that even custom strategies cannot return topics that violate the client's requested pattern.
+
+# Security Considerations
+Input Validation: The properties map is user-supplied input. Implementers of the customizable strategy MUST validate and sanitize these inputs before using them, especially if they are used to construct database queries.
+
+Authorization: This PIP only controls the discovery (listing) of topics. It does not bypass the Authorization Service for consuming or producing to those topics.
+
Review Comment:
Commenting on my own question: I guess topic policy contains the properties at the moment? I guess that PIP-452 only makes sense when using a custom TopicPoliciesService implementation configured with `topicPoliciesServiceClassName`?
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734564690
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
Review Comment:
`Optional.empty()` is invalid result, proceed to the next interceptor or fall back to broker's default logic to list all topics
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2685112467
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
> default CompletableFuture<Optional> interceptGetTopicsOfNamespace(
Applied this suggestion
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689686926
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
Review Comment:
Topic watch is not supported first.
If you want to use this feature, you need to disable topic wach.
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684949182
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
Since we allow the configuration of multiple interceptors, the return here means that the result of current interception is not processed this method. If all the interceptors do not process the method, the default logic will be used.
In addition, the interception configuration is empty by default. In this case, we still need to fall back to the default logic.
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684949182
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
Since we allow the configuration of multiple interceptors, the return here means that the result of this interception is not processed by the later interceptor. If all the interceptors do not process the method, the default logic will be used.
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689489077
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
Review Comment:
How about topic list watchers? There are `CommandWatchTopicList`, `CommandWatchTopicListSuccess` and `CommandWatchTopicUpdate` which would require properties support as well so that the solution would be consistent.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734548551
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
Review Comment:
> ## Out of Scope
>Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
>If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2685446889
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
Review Comment:
Spelling/grammar: The sentence ends with "list all topic" but should be "list all topics" (plural).
```suggestion
* If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topics.
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
Review Comment:
Inconsistent quote style: The CLI example uses a double quote at the end but no opening quote. Either remove the trailing quote or add both opening and closing quotes around the entire command for consistency.
```suggestion
pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
Review Comment:
Incorrect Javadoc reference syntax: The Javadoc reference `{@code Optional#empty}` should be `{@code Optional.empty()}` or `{@link Optional#empty()}` to follow proper Javadoc conventions. The hash symbol is correct for linking to methods, but when using @code, it should show the actual method call syntax with parentheses, or use @link for proper cross-referencing.
```suggestion
* If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topic.
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.getTopics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.getTopics();
+ boolean hasAppliedFilter = listingResult.isFiltered();
+
+ // Apply regex filtering (if the interceptor did not filter and regex matching is enabled)
+ if (!isAlreadyFiltered && enableSubscriptionPatternEvaluation
Review Comment:
Undefined variable: The variable `isAlreadyFiltered` is used at line 166 but is never defined in the method signature or the code shown. This should likely be `hasAppliedFilter` which is defined at line 163.
```suggestion
if (!hasAppliedFilter && enableSubscriptionPatternEvaluation
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
Review Comment:
Inconsistent wrapping behavior: The code shows wrapping the default namespace service result in `TopicListingResult(topics, false)` at line 152, but the interceptor is expected to return `Optional<TopicListingResult>`. The logic then unwraps this at line 144-154, but the code flow shows that line 152 returns a `TopicListingResult` directly, not wrapped in an Optional. This should return `Optional.of(new TopicListingResult(topics, false))` to maintain type consistency.
```suggestion
// Wrap List<String> into TopicListingResult(topics, filtered=false) inside an Optional for unified processing
return getBrokerService().pulsar().getNamespaceService()
.getListOfUserTopics(namespaceName, mode)
.thenApply(topics -> Optional.of(new TopicListingResult(topics, false)));
}
}).thenCompose(listingResultOpt -> {
TopicListingResult listingResult = listingResultOpt.get();
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.getTopics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.getTopics();
Review Comment:
Missing method accessor: The code calls `listingResult.getTopics()`, but `TopicListingResult` is defined as a Java record with a field named `topics` (line 73). For Java records, the accessor method should be `topics()` not `getTopics()`.
```suggestion
long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
listSizeHolder.updateSize(actualSize);
return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
isPermitRequestCancelled, permits -> {
List<String> finalTopics = listingResult.topics();
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
Review Comment:
Type inconsistency in chain: The return type handling is problematic. Line 142 shows `interceptorFuture.thenCompose(interceptorResult -> {...})` where interceptorResult is of type `Optional<TopicListingResult>`. Lines 146 and 152 appear to return different types - line 146 returns `CompletableFuture.completedFuture(interceptorResult)` (a `CompletableFuture<Optional<TopicListingResult>>`), while line 152 returns a `TopicListingResult` directly without Optional wrapping. Then line 154 expects `listingResult` to be a `TopicListingResult`. This type chain is broken and needs to be fixed.
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
Review Comment:
Missing import statement: The code snippet references `Map` interface but doesn't include the required import statement. Add `import java.util.Map;` to the imports section.
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
Review Comment:
Unclear API design: The REST API format shows `properties=k1=v1,k2=v2` at line 54, but it's not clear how this query parameter should be parsed. Is it a single parameter with comma-separated key-value pairs, or multiple parameters? The CLI example at line 59 shows `-p k1=v1 -p k2=v2` which suggests multiple parameters. The REST API documentation should clarify the expected format and whether URL encoding is required for values.
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.getTopics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.getTopics();
Review Comment:
Missing method accessor: The code at line 156 calls `listingResult.getTopics()`, but `TopicListingResult` is defined as a Java record with a field named `topics` (line 73). For Java records, the accessor method should be `topics()` not `getTopics()`. The same issue occurs at line 162.
```suggestion
long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
listSizeHolder.updateSize(actualSize);
return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
isPermitRequestCancelled, permits -> {
List<String> finalTopics = listingResult.topics();
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional#empty}, proceed to the next interceptor or broker's default logic to list all topic.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.getTopics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.getTopics();
+ boolean hasAppliedFilter = listingResult.isFiltered();
Review Comment:
Missing method accessor: The code calls `listingResult.isFiltered()`, but `TopicListingResult` is defined as a Java record with a field named `filtered` (line 73). For Java records, the accessor method should be `filtered()` not `isFiltered()`. Note that for boolean fields in records, the accessor is simply the field name, not prefixed with 'is' or 'get'.
```suggestion
boolean hasAppliedFilter = listingResult.filtered();
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,241 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
Review Comment:
Grammar issue: The parenthetical remark should be clearer. Consider revising to "These properties may be related to topic properties" or better yet, clarify what type of properties are being referenced and their relationship to topic properties.
```suggestion
No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from, topic properties).
```
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2684949182
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
Since we allow the configuration of multiple interceptors, the return here means that the result of current interception is not processed this method. If all the interceptors do not process the method, the default logic will be used.
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2685113670
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
+ }
+
+ /**
+ * Return a successful result with the provided topics list.
+ */
+ public static TopicListingResult success(List<String> topics, boolean filtered) {
+ if (topics == null) {
+ throw new IllegalArgumentException("Topics list cannot be null in success result");
+ }
+ return new TopicListingResult(topics, filtered);
+ }
+
+ /**
+ * Check if this result is a "Pass Through" result.
+ */
+ public boolean isPassThrough() {
+ return this == PASS_THROUGH_INSTANCE;
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * <ul>
+ * <li>{@link TopicListingResult#passThrough()} :
+ * (Default) Proceed to the next interceptor or Broker's default logic.</li>
+ * <li>{@link TopicListingResult#success(java.util.List, boolean)} :
+ * Stop the chain and return these topics immediately.</li>
+ * </ul>
+ * If the future completes with {@code null}, it is treated as {@code passThrough()}.
+ */
+ default CompletableFuture<TopicListingResult> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(TopicListingResult.passThrough());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor
+ CompletableFuture<TopicListingResult> interceptorFuture =
+ getBrokerService().getPulsar().getBrokerInterceptor()
+ .interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && !interceptorResult.isPassThrough()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult);
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> TopicListingResult.success(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ List<String> rawTopics = listingResult.getTopics();
Review Comment:
Simplified pseudocode
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2685112467
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
> default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
Applied this suggestion
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
> default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace()
Applied this suggestion
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689967877
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.topics();
+ boolean isAlreadyFiltered = listingResult.filtered();
+
+ // Apply regex filtering (if the interceptor did not filter and regex matching is enabled)
+ if (!isAlreadyFiltered && enableSubscriptionPatternEvaluation
+ && topicsPattern.isPresent()) {
+ ......
+ }
+
+ // Calculate the Hash and send the response
+ String hash = TopicList.calculateHash(finalTopics);
+ boolean hashUnchanged = topicsHash.isPresent() && topicsHash.get().equals(hash);
+ ......
+ },
+ // Handle logic for heap memory limit exceeded inside withUpdatedPermits
+ t -> {
+ ......
+ return CompletableFuture.completedFuture(null);
+ });
+ }).whenComplete((__, ___) -> {
+ lookupSemaphore.release();
+ }).exceptionally(ex -> {
+ ......
+ return null;
+ });
+ },
+ // Handle logic for initial acquire failure
+ t -> {
+ ......
+ });
+ });
+}
+```
+
+### Client Changes
+Update LookupService
+The internal LookupService interface is updated to accept the properties map.
+
+```Java
+CompletableFuture<List<String>> getTopicsUnderNamespace(
+ NamespaceName namespace,
+ Mode mode,
+ String topicsPattern,
+ String topicsHash,
+ Map<String, String> properties
+);
+```
+Update `PatternMultiTopicsConsumerImpl`
+The regex consumer implementation (PatternMultiTopicsConsumerImpl) will be updated to extract the properties from ConsumerConfigurationData and pass them to the LookupService.
+
+```Java
+// In PatternMultiTopicsConsumerImpl.java
+ConsumerConfigurationData conf;
+Map<String, String> contextProperties = conf.getProperties();
+
+lookup.getTopicsUnderNamespace(
+ namespace,
+ mode,
+ topicsPattern.pattern(),
+ topicsHash,
+ contextProperties // Pass consumer properties here
+).thenAccept(topics -> {
+ // ... update subscriptions ...
+});
+```
+
+
+# Backward Compatibility
+Protocol: Adding an optional field (properties) to the Protobuf definition is non-breaking. Old clients will not send this field; old brokers will ignore it.
+
+Behavior: The default strategy mimics the current behavior exactly. The Broker retains the final filtering logic, ensuring that even custom strategies cannot return topics that violate the client's requested pattern.
+
+# Security Considerations
+Input Validation: The properties map is user-supplied input. Implementers of the customizable strategy MUST validate and sanitize these inputs before using them, especially if they are used to construct database queries.
+
+Authorization: This PIP only controls the discovery (listing) of topics. It does not bypass the Authorization Service for consuming or producing to those topics.
+
Review Comment:
This will be maintained by the implementer of the customizable topic listing plugin, and the relationship between properties and topic does not need to be maintained in pulsar.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2749702290
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
Review Comment:
I think the current definition is more friendly to cli.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734498120
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
Review Comment:
```suggestion
pulsar-admin topics list <tenant>/<namespace> -p k1=v1,k2=v2
```
Just make it consistent with the REST API
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
Review Comment:
```suggestion
default CompletableFuture<Optional<TopicListingResult>> onGetTopicsOfNamespace(
```
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
Review Comment:
It seems incorrect for the interceptor implementation that intentionally to return an empty topic list to the client, for example, no properties match.
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
Review Comment:
We need to keep the pulsar's implementation by default? Otherwise, this change will introduce compatibility issues if user has their own broker interceptor plugin but no implementation for this method.
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
Review Comment:
It seems to be "&"? the topic has both label k1=v1 and k2=v2? Or we need to mention Pulsar will not define this behavior, it depends on the plugin implementation?
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
Review Comment:
How this configuration will be used? Here defined the configuration name, but there is no detailed behavior definition for this configuration.
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2734560755
##########
pip/pip-452.md:
##########
@@ -0,0 +1,263 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+
+## In Scope
+
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+Configuration: Add a broker configuration to enable/disable topic list watching.
+
+## Out of Scope
+
+Not support topic list watching for customized topic listing in this PIP. It can be considered in future work.
+If you want to use this feature, you need to disable topic list watcher by set `enableBrokerTopicListWatcher = false`.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+broker.conf
+```properties
+# Enables watching topic add/remove events on broker side. It is separated from enableBrokerSideSubscriptionPatternEvaluation.
+enableBrokerTopicListWatcher = true
+```
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
Review Comment:
return an empty topic list?
Do you mean that return `Optional.empty()`? `Optional.empty` is invalid result, It means that the interceptor did not handle this request.
https://github.com/apache/pulsar/pull/25134#discussion_r2684569996
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689919155
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+Admin API & CLI: Update the REST API and CLI to accept properties for listing topics in a namespace.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API (URL encoding is required for values):
+```
+GET /admin/v2/persistent/{tenant}/{namespace}?properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+import java.util.Objects;
+
+public record TopicListingResult (List<String> topics, boolean filtered) {
+
+ public TopicListingResult{
+ Objects.requireNonNull(topics, "topics");
+ }
+}
+```
+Update BrokerInterceptor Interface
+Add a default method in BrokerInterceptor. It returns null by default, indicating no interception (to maintain backward compatibility).
+
+```Java
+
+package org.apache.pulsar.broker.intercept;
+
+import java.util.Optional;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.concurrent.CompletableFuture;
+import java.util.Map;
+
+public interface BrokerInterceptor extends AutoCloseable {
+
+ // ... existing methods ...
+
+ /**
+ * Intercept the GetTopicsOfNamespace request.
+ * <p>
+ * This method allows plugins to override the default topic discovery logic (ZooKeeper scan).
+ * It enables fetching topics from external sources (e.g., databases, other metadata stores)
+ * based on the provided client context properties.
+ *
+ * @param namespace The namespace being queried.
+ * @param mode The query mode (PERSISTENT, NON_PERSISTENT, or ALL).
+ * @param topicsPattern Optional regex pattern provided by the client.
+ * @param properties Context properties provided by the client.
+ * @return A CompletableFuture containing the result:
+ * If the future completes with {@code Optional.empty()}, proceed to the next interceptor or broker's default logic to list all topics.
+ */
+ default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace(
+ NamespaceName namespace,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern,
+ Map<String, String> properties) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+}
+```
+
+ServerCnx Logic Update
+In ServerCnx.java, the method internalHandleGetTopicsOfNamespace will be updated to use this strategy.
+
+```java
+private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName namespaceName, long requestId,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String> topicsPattern, Optional<String> topicsHash,
+ Map<String, String> properties,
+ Semaphore lookupSemaphore) {
+ BooleanSupplier isPermitRequestCancelled = () -> !ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder listSizeHolder =
+ service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(), mode);
+
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled, initialPermits -> {
+ // Call the Interceptor if exists
+ BrokerInterceptor brokerInterceptor = getBrokerService().getPulsar().getBrokerInterceptor();
+ CompletableFuture<Optional<TopicListingResult>> interceptorFuture =
+ brokerInterceptor == null ? CompletableFuture.completedFuture(Optional.empty()) :
+ brokerInterceptor.interceptGetTopicsOfNamespace(namespaceName, mode, topicsPattern, properties);
+
+ return interceptorFuture.thenCompose(interceptorResult -> {
+ // Decision branch: Use the interceptor result OR fall back to the default logic
+ if (interceptorResult != null && interceptorResult.isPresent()) {
+ // case A: The interceptor has taken over the request
+ return CompletableFuture.completedFuture(interceptorResult.get());
+ } else {
+ // case B: The interceptor did not handle, so the original query logic is executed.
+ // Wrap List<String> into TopicListingResult(topics, filtered=false) for unified processing
+ return getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenApply(topics -> new TopicListingResult(topics, false));
+ }
+ }).thenCompose(listingResult -> {
+ // Adjust memory permits based on ACTUAL size
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(listingResult.topics());
+ listSizeHolder.updateSize(actualSize);
+
+ return maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
+ isPermitRequestCancelled, permits -> {
+
+ List<String> finalTopics = listingResult.topics();
+ boolean isAlreadyFiltered = listingResult.filtered();
+
+ // Apply regex filtering (if the interceptor did not filter and regex matching is enabled)
+ if (!isAlreadyFiltered && enableSubscriptionPatternEvaluation
+ && topicsPattern.isPresent()) {
+ ......
+ }
+
+ // Calculate the Hash and send the response
+ String hash = TopicList.calculateHash(finalTopics);
+ boolean hashUnchanged = topicsHash.isPresent() && topicsHash.get().equals(hash);
+ ......
+ },
+ // Handle logic for heap memory limit exceeded inside withUpdatedPermits
+ t -> {
+ ......
+ return CompletableFuture.completedFuture(null);
+ });
+ }).whenComplete((__, ___) -> {
+ lookupSemaphore.release();
+ }).exceptionally(ex -> {
+ ......
+ return null;
+ });
+ },
+ // Handle logic for initial acquire failure
+ t -> {
+ ......
+ });
+ });
+}
+```
+
+### Client Changes
+Update LookupService
+The internal LookupService interface is updated to accept the properties map.
+
+```Java
+CompletableFuture<List<String>> getTopicsUnderNamespace(
+ NamespaceName namespace,
+ Mode mode,
+ String topicsPattern,
+ String topicsHash,
+ Map<String, String> properties
+);
+```
+Update `PatternMultiTopicsConsumerImpl`
+The regex consumer implementation (PatternMultiTopicsConsumerImpl) will be updated to extract the properties from ConsumerConfigurationData and pass them to the LookupService.
+
+```Java
+// In PatternMultiTopicsConsumerImpl.java
+ConsumerConfigurationData conf;
+Map<String, String> contextProperties = conf.getProperties();
+
+lookup.getTopicsUnderNamespace(
+ namespace,
+ mode,
+ topicsPattern.pattern(),
+ topicsHash,
+ contextProperties // Pass consumer properties here
+).thenAccept(topics -> {
+ // ... update subscriptions ...
+});
+```
+
+
+# Backward Compatibility
+Protocol: Adding an optional field (properties) to the Protobuf definition is non-breaking. Old clients will not send this field; old brokers will ignore it.
+
+Behavior: The default strategy mimics the current behavior exactly. The Broker retains the final filtering logic, ensuring that even custom strategies cannot return topics that violate the client's requested pattern.
+
+# Security Considerations
+Input Validation: The properties map is user-supplied input. Implementers of the customizable strategy MUST validate and sanitize these inputs before using them, especially if they are used to construct database queries.
+
+Authorization: This PIP only controls the discovery (listing) of topics. It does not bypass the Authorization Service for consuming or producing to those topics.
+
Review Comment:
TopicPolicies (https://github.com/apache/pulsar/blob/master/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java#L46) doesn't contain the properties. How would the properties be added, modified and deleted for topics?
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2685112467
##########
pip/pip-452.md:
##########
@@ -0,0 +1,306 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Extend `BrokerInterceptor` to allow plugins to intercept the "Get Topics" request.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will add a new method to the BrokerInterceptor interface.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Protocol Changes
+Update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+
+### REST API & CLI Changes
+
+Add `properties` parameter to the REST API endpoint for listing topics in a namespace to list topic with specific properties for customizable topic listing.
+
+REST API:
+```
+GET /admin/v2/persistent/{tenant}/{namespace} with a query params properties=k1=v1,k2=v2
+```
+
+CLI:
+```
+pulsar-admin topics list <tenant>/<namespace> -p k1=v1 -p k2=v2"
+```
+
+## Design & Implementation Details
+### Broker Changes
+
+Add `TopicListingResult` for listing topic results
+
+```java
+package org.apache.pulsar.broker.intercept;
+
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class TopicListingResult {
+
+ // Define a singleton "Pass Through" object
+ private static final TopicListingResult PASS_THROUGH_INSTANCE = new TopicListingResult(null, false);
+
+ private final List<String> topics;
+ private final boolean filtered;
+
+ private TopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+
+ /**
+ * Return an invalid result indicating that the caller should continue with the default logic.
+ */
+ public static TopicListingResult passThrough() {
+ return PASS_THROUGH_INSTANCE;
Review Comment:
> `default CompletableFuture<Optional<TopicListingResult>> interceptGetTopicsOfNamespace()`
Applied this suggestion
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2681636944
##########
pip/pip-452.md:
##########
@@ -0,0 +1,212 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Introduce a `NamespaceTopicListingStrategy` SPI to customize how topics are retrieved.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will abstract the topic retrieval logic into a NamespaceTopicListingStrategy.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+1. Protocol Changes
+We update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+2. Broker Changes
+New Interface: `NamespaceTopicListingStrategy`
+We introduce a strategy interface in the broker.
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+
+public class NamespaceTopicListingResult {
+ private final List<String> topics;
+
+ // Indicates if the plugin has already applied the regex filtering provided in the request.
+ // If true, the Broker will SKIP the subsequent filtering step.
+ private final boolean filtered;
+
+ public NamespaceTopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+ public List<String> getTopics() {
+ return topics;
+ }
+
+ public boolean isFiltered() {
+ return filtered;
+ }
+}
+```
+
+```Java
+
+package org.apache.pulsar.broker.namespace;
+
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface NamespaceTopicListingStrategy {
Review Comment:
Applied the suggestions
--
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] [feat][broker] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2689775958
##########
pip/pip-452.md:
##########
@@ -0,0 +1,245 @@
+# PIP-452: Customizable topic listing of namespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (these client properties may correspond to, or be derived from topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
Review Comment:
Sure, I update doc and add configuration `enableBrokerTopicListWatcher`
--
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] [feat][pip] PIP-452: Customizable topic listing of namespace with properties [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#issuecomment-3871872289
Since the vote has been passed, merge this PIP
--
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] [feat][broker] PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #25134:
URL: https://github.com/apache/pulsar/pull/25134#discussion_r2680899806
##########
pip/pip-452.md:
##########
@@ -0,0 +1,212 @@
+# PIP-452: Customizable Topic Listing in GetTopicsOfNamespace with properties
+
+# Motivation
+Currently, the CommandGetTopicsOfNamespace logic in the Pulsar Broker is hard-coded to scan the metadata store (ZooKeeper) for all children nodes under a namespace.
+
+This implementation limits the flexibility required for complex multi-tenant scenarios:
+
+No Client Context: The broker cannot distinguish who is asking for the topics or why. It cannot filter topics based on client properties (This properties may be related to topic properties).
+
+Inefficient Filtering: For namespaces with millions of topics, the broker must fetch the full list into memory before applying the topics_pattern regex. There is no way to "push down" the filtering to the data source (e.g., a database with an index).
+
+To address these issues, I propose making the topic listing logic pluggable and extending the protocol to accept client properties.
+
+# Goals
+Protocol: Add a properties field to `CommandGetTopicsOfNamespace` to carry client-side context.
+
+Broker: Introduce a `NamespaceTopicListingStrategy` SPI to customize how topics are retrieved.
+
+Client: Update the Java Client to forward Consumer properties to the lookup service when using Regex subscriptions.
+
+# High Level Design
+We will modify the Pulsar Protocol to carry a properties map. On the Broker side, we will abstract the topic retrieval logic into a NamespaceTopicListingStrategy.
+The default implementation will preserve the existing behavior (fetching from ZooKeeper). The Broker's connection handler will simply delegate the request to this strategy.
+
+# Detailed Design
+1. Protocol Changes
+We update `PulsarApi.proto` to include the properties field.
+
+PulsarApi.proto
+```protobuf
+message CommandGetTopicsOfNamespace {
+ required uint64 request_id = 1;
+ required string namespace = 2;
+ optional Mode mode = 3 [default = PERSISTENT];
+
+ // Existing fields for filtering and hash optimization
+ optional string topics_pattern = 4;
+ optional string topics_hash = 5;
+
+ // New field: Context properties from the client
+ repeated KeyValue properties = 6;
+}
+```
+2. Broker Changes
+New Interface: `NamespaceTopicListingStrategy`
+We introduce a strategy interface in the broker.
+
+```java
+package org.apache.pulsar.broker.namespace;
+
+import java.util.List;
+
+public class NamespaceTopicListingResult {
+ private final List<String> topics;
+
+ // Indicates if the plugin has already applied the regex filtering provided in the request.
+ // If true, the Broker will SKIP the subsequent filtering step.
+ private final boolean filtered;
+
+ public NamespaceTopicListingResult(List<String> topics, boolean filtered) {
+ this.topics = topics;
+ this.filtered = filtered;
+ }
+
+ public List<String> getTopics() {
+ return topics;
+ }
+
+ public boolean isFiltered() {
+ return filtered;
+ }
+}
+```
+
+```Java
+
+package org.apache.pulsar.broker.namespace;
+
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public interface NamespaceTopicListingStrategy {
Review Comment:
It's better to extend the existing interceptor rather than introducing yet another plugin, i.e. add a new method to the existing interface rather than introducing a new config and a new interface
--
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