[fix][broker] PIP-468: Fix listener leak in DagWatchSession (#25650)

Author: merlimat

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

@@ -75,13 +75,26 @@ public class ScalableTopicResources extends BaseResources<ScalableTopicMetadata>
     private final MetadataCache<ConsumerRegistration> consumerRegistrationCache;
 
     /**
-     * Per-namespace listeners for scalable topic create / modify / delete events.
-     * Keyed by listener so each subscriber can deregister cleanly on close. The map
-     * is consulted from the single store-level listener registered at construction
-     * time, eliminating the listener leak that would otherwise occur every time a
-     * watcher session ends (the metadata store API has no
-     * {@code unregisterListener}). Mirrors the {@link TopicResources}
-     * pattern for {@code TopicListener}.
+     * Per-path listeners for scalable-topic metadata events. Each listener watches a
+     * single exact path (typically a topic record); the resources-level fan-out
+     * dispatches notifications whose path equals the listener's registered path.
+     * Used by {@link DagWatchSession}-style subscribers that want events for one
+     * specific topic.
+     *
+     * <p>Hosted here — rather than letting each subscriber call
+     * {@code store.registerListener} directly — because {@code MetadataStore} has no
+     * {@code unregisterListener}: per-subscriber direct registration would leak a
+     * listener for the broker's lifetime every time a session ends, and every
+     * metadata notification would fan out to all stale listeners. Mirrors
+     * {@link TopicResources} for {@code TopicListener}.
+     */
+    private final Map<MetadataPathListener, String> pathListeners = new ConcurrentHashMap<>();
+
+    /**
+     * Per-namespace listeners for scalable-topic create / modify / delete events.
+     * Used by namespace-wide watchers (e.g. multi-topic consumer wrappers); the
+     * fan-out matches direct children of the listener's namespace base path. Same
+     * leak-avoidance rationale as {@link #pathListeners}.
      */
     private final Map<NamespaceListener, NamespaceName> namespaceListeners =
             new ConcurrentHashMap<>();
@@ -90,16 +103,47 @@ public ScalableTopicResources(MetadataStore store, int operationTimeoutSec) {
         super(store, ScalableTopicMetadata.class, operationTimeoutSec);
         this.subscriptionCache = store.getMetadataCache(SubscriptionMetadata.class);
         this.consumerRegistrationCache = store.getMetadataCache(ConsumerRegistration.class);
-        // Single shared metadata-store listener fans out to every registered watcher.
-        // Per-watcher registration happens via registerNamespaceListener; close() calls
-        // deregisterNamespaceListener so closed watchers are not on the dispatch list.
+        // Single shared metadata-store listener fans out to both per-path and
+        // per-namespace subscribers. Per-subscriber lifecycle goes through the
+        // register / deregister methods below.
         if (store instanceof MetadataStoreExtended ext) {
             ext.registerListener(this::handleNotification);
         } else {
             store.registerListener(this::handleNotification);
         }
     }
 
+    // --- Per-path metadata listeners ---
+
+    /**
+     * Listener for metadata events on a specific scalable-topic-related path. The
+     * fan-out in {@link ScalableTopicResources} compares each notification's path
+     * against {@link #getMetadataPath()} and dispatches on exact match.
+     */
+    public interface MetadataPathListener {
+        /** Exact path this listener is interested in (no wildcard / prefix). */
+        String getMetadataPath();
+
+        /** Called for every metadata event on the listener's path. */
+        void onNotification(Notification notification);
+    }
+
+    /**
+     * Register a per-path metadata listener. Idempotent — re-registering the same
+     * listener just refreshes its path mapping (e.g. if the listener moved its path).
+     */
+    public void registerPathListener(MetadataPathListener listener) {
+        pathListeners.put(listener, listener.getMetadataPath());
+    }
+
+    /**
+     * Deregister a previously-registered listener. Safe to call multiple times or for
+     * listeners that were never registered.
+     */
+    public void deregisterPathListener(MetadataPathListener listener) {
+        pathListeners.remove(listener);
+    }
+
     // --- Namespace-level scalable-topics listeners ---
 
     /**
@@ -135,35 +179,53 @@ public void deregisterNamespaceListener(NamespaceListener listener) {
     }
 
     /**
-     * Single fan-out path: for each registered listener, emit the notification iff
-     * its path is a direct child of the listener's namespace base path. Filters out
-     * subtree events (subscriptions, controller lock) up front.
+     * Single fan-out path. For each registered subscriber:
+     * <ul>
+     *   <li>Path listener: dispatch when the notification's path equals the listener's
+     *       registered path.</li>
+     *   <li>Namespace listener: dispatch when the notification's path is a direct
+     *       child of {@code /topics/<tenant>/<ns>} (skips subtree events like
+     *       subscriptions / controller lock).</li>
+     * </ul>
      */
     void handleNotification(Notification notification) {
-        if (namespaceListeners.isEmpty()) {
-            return;
-        }
         String path = notification.getPath();
-        if (!path.startsWith(SCALABLE_TOPIC_PATH + "/")) {
-            return;
-        }
-        for (Map.Entry<NamespaceListener, NamespaceName> entry : namespaceListeners.entrySet()) {
-            String basePath = namespacePath(entry.getValue());
-            if (!path.startsWith(basePath + "/")) {
-                continue;
-            }
-            // Direct child only — strip the prefix and check there's no further '/'.
-            String rest = path.substring(basePath.length() + 1);
-            if (rest.indexOf('/') >= 0) {
-                continue;
+
+        // Path listeners — exact match.
+        if (!pathListeners.isEmpty()) {
+            for (Map.Entry<MetadataPathListener, String> entry : pathListeners.entrySet()) {
+                if (entry.getValue().equals(path)) {
+                    try {
+                        entry.getKey().onNotification(notification);
+                    } catch (Exception e) {
+                        log.warn().attr("listener", entry.getKey())
+                                .attr("path", path)
+                                .exceptionMessage(e)
+                                .log("Failed to dispatch scalable-topic path notification");
+                    }
+                }
             }
-            try {
-                entry.getKey().onNotification(notification);
-            } catch (Exception e) {
-                log.warn().attr("listener", entry.getKey())
-                        .attr("path", path)
-                        .exceptionMessage(e)
-                        .log("Failed to dispatch scalable-topic notification");
+        }
+
+        // Namespace listeners — direct child of /topics/<ns>.
+        if (!namespaceListeners.isEmpty() && path.startsWith(SCALABLE_TOPIC_PATH + "/")) {
+            for (Map.Entry<NamespaceListener, NamespaceName> entry : namespaceListeners.entrySet()) {
+                String basePath = namespacePath(entry.getValue());
+                if (!path.startsWith(basePath + "/")) {
+                    continue;
+                }
+                String rest = path.substring(basePath.length() + 1);
+                if (rest.indexOf('/') >= 0) {
+                    continue;
+                }
+                try {
+                    entry.getKey().onNotification(notification);
+                } catch (Exception e) {
+                    log.warn().attr("listener", entry.getKey())
+                            .attr("path", path)
+                            .exceptionMessage(e)
+                            .log("Failed to dispatch scalable-topic namespace notification");
+                }
             }
         }
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/DagWatchSession.java

@@ -48,7 +48,7 @@
  * <p>The session is tied to a connection. When the connection breaks, the session dies.
  * The client must reinitiate a new session (possibly with another broker).
  */
-public class DagWatchSession {
+public class DagWatchSession implements ScalableTopicResources.MetadataPathListener {
 
     private static final Logger LOG = Logger.get(DagWatchSession.class);
     private final Logger log;
@@ -61,7 +61,6 @@ public class DagWatchSession {
     private final BrokerService brokerService;
 
     private final String metadataPath;
-    private final java.util.function.Consumer<Notification> notificationListener;
     private volatile boolean closed = false;
 
     public DagWatchSession(long sessionId,
@@ -75,17 +74,22 @@ public DagWatchSession(long sessionId,
         this.resources = resources;
         this.brokerService = brokerService;
         this.metadataPath = resources.topicPath(topicName);
-        this.notificationListener = this::onNotification;
         this.log = LOG.with().attr("topic", topicName).attr("sessionId", sessionId).build();
     }
 
+    @Override
+    public String getMetadataPath() {
+        return metadataPath;
+    }
+
     /**
      * Start the session: load current metadata, set up watch, and return
      * the initial layout response.
      */
     public CompletableFuture<ScalableTopicLayoutResponse> start() {
-        // Register metadata store listener for changes to this topic's metadata
-        resources.getStore().registerListener(notificationListener);
+        // Register through the resources-level fan-out so close() can deregister us
+        // and we don't accumulate stale store-level listeners over time.
+        resources.registerPathListener(this);
 
         return resources.getScalableTopicMetadataAsync(topicName, true)
                 .thenCompose(optMd -> {
@@ -98,8 +102,13 @@ public CompletableFuture<ScalableTopicLayoutResponse> start() {
                 });
     }
 
-    // Visible for testing — invoked by the metadata-store listener registered in start().
-    void onNotification(Notification notification) {
+    /**
+     * Invoked by the {@link ScalableTopicResources} fan-out for every metadata event
+     * matching this session's topic path. The registry already path-filtered for us;
+     * we re-check defensively so a registry-level bug can't cause a reload storm.
+     */
+    @Override
+    public void onNotification(Notification notification) {
         if (closed) {
             return;
         }
@@ -184,7 +193,10 @@ private ScalableTopicDAG buildDagProto(ScalableTopicLayoutResponse response) {
 
     public void close() {
         closed = true;
-        // Listener is guarded by the closed flag; MetadataStore does not support unregister.
+        // Drop ourselves from the resources' fan-out so the per-event dispatch skips
+        // us — no listener leak, no per-notification dispatch tax across the broker's
+        // lifetime.
+        resources.deregisterPathListener(this);
     }
 
     /**