[feat][broker] PIP-451: Support Label-based Topic Subscription

pip/pip-451.md

@@ -0,0 +1,255 @@
+# PIP-451: Support Label-based Topic Subscription
+
+# Motivation
+
+Currently, Apache Pulsar supports Pattern Subscription, which allows consumers to subscribe to multiple topics using Regular Expressions (Regex). While powerful, Regex-based subscription has several structural limitations in complex microservice architectures:
+
+* Coupling: It couples the consumption logic with the topic naming convention. Changing business requirements often forces topic renaming, which is operationally expensive and risky.S
+
+* Flexibility: It is difficult to group semantically related but differently named topics (e.g., persistent://public/default/payment-core and persistent://public/legacy/billing-v1) into a single subscription without complex regex wizardry.
+
+* Complexity: complex Regex can be hard to maintain and error-prone.
+
+Label-based Subscription solves these issues by decoupling "Identity" (Topic Name) from "Attributes" (Labels). Users can attach Key-Value metadata (e.g., env=prod, dept=finance) to topics and subscribe by specifying a label selector.
+
+# Goals
+
+## In Scope
+* Management: Allow attaching, updating, and removing Key-Value labels to/from Topics via the Admin API.
+* Subscription: Allow Consumers to subscribe to topics matching specific Labels within specified Namespaces. Support cross-namespace subscription via an explicit namespace list in the Client API, avoiding the complexity of background metadata polling.
+
+
+
+# High-Level Design
+The design introduces a metadata-driven approach where labels are stored in TopicPolicies. 
+The Broker maintains an in-memory index to map labels to topics. The client utilizes a "Watch" mechanism to receive real-time updates when topics matching the labels are created or updated.
+## Key points
+* Storage: Labels are stored as Map<String, String> inside TopicPolicies.
+* Indexing: The Broker maintains an In-Memory Inverted Index (LabelKey -> LabelValue -> Set<Topic>) per Namespace. This hierarchical structure ensures efficient lookups for Key-Value pairs without iterating through all topics.
+* Discovery Protocol: We extend the CommandWatchTopicList protocol (PIP-179) to accept a label_selector.
+* Client Implementation: The Client accepts a list of target namespaces and manages multiple watchers (one per namespace) to aggregate matching topics.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Storage 
+
+#### Topic Labels in TopicPolicies
+
+Add a labels field to the TopicPolicies class. Since Topic Policies are propagated via the __change_events system topic, this ensures durability and consistency across brokers.
+```java
+public class TopicPolicies {
+    // New field: Key-Value labels
+    private Map<String, String> customLabels;
+}
+```
+
+#### In-Memory Inverted Index for Labels-Topic Mapping
+
+The SystemTopicBasedTopicPoliciesService will maintain a nested map structure per Namespace to support efficient Key-Value lookups.
+
+**Data Structure**:
+
+```java
+// Map<Namespace, Map<LabelKey, Map<LabelValue, Set<TopicName>>>>
+Map<String, Map<String, Map<String, Set<String>>>> labelTopicInvertedIndex;
+```
+
+### Topic Labels Observation and Notification Mechanism
+
+### TopicPoliciesService Interface Changes
+
+```java
+public interface TopicPoliciesService {
+    // ... Existing methods ...
+
+    /**
+     *  Register a label change listener with the service.
+     */
+    void registerTopicLabelsListener(NamespaceName namespace, TopicLabelsListener listener);
+
+    /**
+     * Unregister a label change listener with the service.
+     */
+    void unregisterTopicLabelsListener(NamespaceName namespace, TopicLabelsListener listener);
+
+    /**
+     * Query the list of all matching Topics under the specified Namespace based on the label selector.
+     */
+    CompletableFuture<List<TopicName>> getTopicsByLabels(NamespaceName namespace, Map<String, String> labels);
+
+    /**
+     * Query all existing labels within a Namespace.
+     */
+    CompletableFuture<Map<String, Set<String>>> getAllLabels(NamespaceName namespace);
+}
+```
+To support real-time label change notifications, we will introduce a listener mechanism in TopicPoliciesService.
+
+#### Add TopicLabelsListener Interface
+
+```java
+public interface TopicLabelsListener {
+    /**
+     * Triggered when topic labels are substantially changed.
+     * @param topicName The affected topic.
+     * @param allLabels The latest snapshot of all labels.
+     */
+    void onLabelsUpdate(TopicName topicName, Map<String, String> allLabels);
+}
+```
+
+#### Listener Management in TopicPoliciesService
+
+The TopicPoliciesService will maintain a list of TopicLabelsListener and notify them when a policy update includes label changes.
+
+#### Notification Logic in Broker
+* Change Detection: When SystemTopicBasedTopicPoliciesService consumes a policy event, it compares oldLabels from the cache with newLabels from the event, then update in-memory inverted index (labelTopicInvertedIndex).
+
+* Global Dispatch: If `!Objects.equals(oldLabels, newLabels)`, the service invokes `onLabelsUpdate(topicName, labels)` for all registered listeners.
+
+* Watcher Evaluation: The TopicListService (registered as a listener) iterates through its TopicListWatcher instances.
+
+#### TopicListWatcher State Machine
+
+Each TopicListWatcher maintains a set of currently matched topics to perform state-based updates:
+
+```java
+public void onLabelsUpdate(TopicName topicName, @Nullable Map<String, String> allLabels) {
+    // allLabels can be null if the topic is deleted
+    boolean wasMatching = matchedTopics.contains(topicName);
+    boolean matchesNow = allLabels != null && labelMatcher.matches(allLabels);
+
+    List<String> newTopics = Collections.emptyList();
+    List<String> deletedTopics = Collections.emptyList();
+
+    if (!wasMatching && matchesNow) {
+        // It did not match before, but now it matches (matches after adding or modifying labels).
+        newTopics = Collections.singletonList(topicName.toString());
+        matchedTopics.add(topicName.toString());
+    } else if (wasMatching && !matchesNow) {
+        //  It matched before, but now it no longer matches (fails to match after the Topic is deleted or its labels are modified).
+        deletedTopics = Collections.singletonList(topicName.toString());
+        matchedTopics.remove(topicName.toString());
+    } else {
+        // If the state remains unchanged
+        return;
+    }
+
+    String hash = TopicList.calculateHash(matchedTopics);
+    topicListService.sendTopicListUpdate(id, hash, deletedTopics, newTopics);
+}
+```
+
+### Client Implementation Details
+The Client implementation acts as an orchestrator to support the "Cross-Namespace" requirement defined in topicsByLabel.
+
+* Multi-Watcher Orchestration:
+
+When `subscribe()` is called, the Client iterates over the provided namespaces.
+For each namespace, it initiates a CommandWatchTopicList request. 
+
+* Aggregation:
+
+The Client maintains a unified view of matching topics.
+New Topic Event: If any of the namespace watchers receives a NEW_TOPIC update (meaning a topic in that namespace matched the labels), the Client adds it to the list and creates a child consumer.
+Deleted Topic Event: If a watcher receives a DELETED_TOPIC update (topic deleted or label removed), the Client closes the corresponding child consumer.
+
+* Deterministic Scope:
+
+Unlike Regex subscription which might require scanning metadata for matching namespaces, this design relies on the user providing the `Set<String>` namespaces.
+### Public API & CLI Changes
+
+#### Set Custom Labels:
+
+* CLI: pulsar-admin topics set-custom-labels <topic-name> --labels "key1=value1,key2=value2"
+* REST API: POST /admin/v2/topics/{tenant}/{namespace}/{topic}/custom-labels with a JSON payload {"labels": {"key1":"value1", "key2":"value2"}}
+* Action: Sets or updates custom labels for the specified topic. The broker (or admin client before sending) will validate that all provided keys (e.g., key1, key2) are present in the allowedCustomLabelKeys list defined in broker.conf. Invalid keys will result in an error. This operation will update the topic's policy and publish a change event to the system topic (__change_events) for that namespace.
+
+#### Get Custom Labels:
+
+* CLI: pulsar-admin topics get-custom-labels <topic-name>
+* REST API: GET /admin/v2/topics/{tenant}/{namespace}/{topic}/custom-labels
+* Action: Retrieves the currently set custom labels for the topic.
+
+#### Remove Custom Labels:
+
+* CLI: 
+  - pulsar-admin topics remove-custom-labels <topic-name> --labels "key1,key2" (to remove specific labels)
+  - pulsar-admin topics remove-custom-labels <topic-name> --all (to remove all custom labels from the topic)
+* pulsar-admin topics remove-custom-labels <topic-name> --labels "key1,key2" (to remove specific labels)
+* pulsar-admin topics remove-custom-labels <topic-name> --all (to remove all custom labels from the topic)
+* REST API: DELETE /admin/v2/topics/{tenant}/{namespace}/{topic}/custom-labels with a query params keys=k1&keys=k2 or all=true.
+* Action: Removes the specified custom labels or all custom labels from the topic. This also updates the topic policy.
+
+#### Query topic associated with specific labels:
+
+* CLI: pulsar-admin topics list <namespace> --custome-labels "k1:v1,k2:v2"
+* REST API: GET /admin/v2/topics/{tenant}/{namespace} with a query params custome-labels=k1:v1,k2:v2
+
+#### Get all existing labels within a namespace:
+
+* CLI: pulsar-admin namespaces list-topic-custome-labels <namespace>
+* REST API: GET
+           /admin/v2/topics/{tenant}/{namespace}/topicCustomeLabels
+
+### Consumer API (Client)
+Introduce `topicsByLabel` in the ConsumerBuilder.
+```java
+public interface ConsumerBuilder<T> {
+
+    // ... existing methods ...
+
+    /**
+     * Subscribe to topics matching the given labels within specific namespaces.
+     * * @param labels     The Key-Value pairs that a topic must match (AND semantics).
+     * @param namespaces The set of namespaces to look for topics. 
+     * If empty, defaults to the current consumer's namespace.
+     */
+    ConsumerBuilder<T> topicsByLabel(Map<String, String> labels, Set<String> namespaces);
+}
+```
+### Binary protocol
+
+Extend CommandWatchTopicList to include label_selector:
+```protobuf
+message CommandWatchTopicList {
+    required uint64 request_id     = 1;
+    required uint64 watcher_id     = 2;
+    required string namespace      = 3;
+    required string topics_pattern = 4;
+    // Only present when the client reconnects:
+    optional string topics_hash    = 5;
+
+
+    // If not empty, the broker filters topics by these labels.
+    // 'topics_pattern' should be ignored if this field is set.
+    repeated KeyValue labels        = 6;
+}
+```
+
+### Configuration Changes
+
+maxCustomLabelValueLength=<integer>
+Description: The maximum character length for a custom label value.
+Default: 128
+
+# Security Considerations
+Authorization: Setting labels modifies Topic Policies, which requires update-policies permission on the Namespace or Topic.
+
+# Backward Compatibility
+Protocol: Old brokers will not understand the label_selector field in CommandWatchTopicList. Old brokers might ignore the `label_selector` field and return all topics (acting as a simple Watch).
+
+Mitigation: The Client library must check the Broker's protocol version or capabilities. If the Broker does not support Label Subscription, the Client should throw a NotSupportedException rather than falling back to consuming all topics (which would be dangerous).
+
+API: This is a purely additive change to the Public API.
+
+# Alternatives
+Using topic properties to store labels, since we need to reverse query the topic list through labels, simply storing labels in the topic properties is insufficient. This is because it is difficult to collect all properties under a single namespace. Although this issue can be resolved by adding a secondary index of {labels}/{topic-list} in the metadata, it may introduce challenges to data consistency. Therefore, this is not considered a priority implementation solution. 
+
+# Links
+
+
+* Mailing List discussion thread:
+* Mailing List voting thread: