[feat] PIP-468: Namespace scalable-topics watcher (protocol + broker + V5 client)

[merlimat]

Summary

Foundation for multi-topic QueueConsumer / StreamConsumer subscriptions
filtered by topic properties: a long-lived broker → client watch session over
the union of scalable topics in a namespace that match a (possibly empty) set
of property filters. The broker pushes a full Snapshot on subscribe and an
incremental Diff (with adds + removes batched together) when membership
changes. Reliable across reconnects via a hash-skip optimisation — when the
client's hash of its current set matches the broker's freshly-computed one,
the broker stays silent.

This PR lands the wire layer, the broker session, and the V5 client watcher.
The multi-topic consumer wrappers themselves come in a follow-up PR.

Design

Goal

Let QueueConsumer and StreamConsumer subscribe to the union of scalable
topics in a namespace that match a (possibly empty) set of property filters.
The matching set must follow live: when topics enter or leave the filter, the
consumer attaches/detaches automatically. Reliable across reconnects and
broker bounces.

Out of scope: CheckpointConsumer (stays single-topic), partitioned and
non-partitioned legacy topics.

Wire protocol

WATCH_SCALABLE_TOPICS         (76)  client -> broker  open watch
WATCH_SCALABLE_TOPICS_UPDATE  (77)  broker -> client  Snapshot or Diff
WATCH_SCALABLE_TOPICS_CLOSE   (78)  client -> broker  close watch

CommandWatchScalableTopics carries watch_id, namespace,
property_filters, and optional current_hash.

CommandWatchScalableTopicsUpdate carries an event proto oneof:

• Snapshot { topics: [string] } — full set, sent on initial subscribe and
  on reconnect when the hash differs;
• Diff { added: [string], removed: [string] } — incremental changes, with
  apply-removed-first semantics (covers rapid remove-then-add).

Wire size: snapshot bounded by the 5 MB frame limit. Defer pagination —
revisit if a namespace ever exceeds.

Reliability

Resync-on-reconnect, no durable event log on the broker.

1. Subscribe. Broker registers as a per-namespace listener on the shared
   ScalableTopicResources notification fan-out first, computes the initial
   filtered set, emits Snapshot{topics}, populates server-side currentSet.
   Events that arrived during snapshot computation are replayed after via the
   same dedup logic the deltas use.

2. Steady state. Each metadata event is filter-evaluated. If membership
   changes relative to currentSet, broker emits Diff{added, removed} and
   updates currentSet. Server-side coalescing window (~50 ms) batches
   nearby events into one Diff.

3. Disconnect. Broker drops the session and deregisters from the
   ScalableTopicResources listener registry.

4. Reconnect with hash short-circuit. Client maintains a hash over its
   current set (CRC32C of sorted topic names — same function used by
   CommandGetTopicsOfNamespace). On reconnect it re-opens the watch and
   passes that hash via CommandWatchScalableTopics.current_hash. The broker
   computes the same hash over its freshly evaluated set:

   • Hash matches → broker emits nothing. The watch is live, the
     client's local state is correct, future deltas flow as usual.
   • Hash differs → broker emits a fresh Snapshot. Client reconciles
     locally — anything in the new snapshot it didn't know about → open
     per-topic consumer; anything it knew but missing → close + flush.

   First subscribe sends an empty / absent hash, which the broker treats as
   "no prior state" and unconditionally emits the initial Snapshot. The
   reconnect-backoff reset is keyed off write success of the watch frame, so
   the hash-skip path (no inbound Snapshot) doesn't keep the backoff at its
   peak across successive short blips.

Properties: idempotent (snapshot is full-set replace; diffs are set ops);
self-healing across any disconnect duration; no broker affinity (any broker
can serve, every broker has the same metadata events). For the common
short-blip reconnect where membership didn't change, the wire cost collapses
to one inbound WatchScalableTopics frame and zero outbound.

Filter evaluation: broker-side

Initial set is computed via the existing findScalableTopicsByPropertiesAsync.
Each metadata event:

• Created / Modified: read the new value, evaluate filter, emit Diff if
  membership changed vs currentSet.
• Deleted: no new value; emit Removed if topic was in currentSet.

Cost: one filter evaluation per metadata event per watcher. Filters are tiny;
fine.

Listener lifecycle

ScalableTopicResources registers a single shared metadata-store listener at
construction and exposes registerNamespaceListener /
deregisterNamespaceListener to per-watcher subscribers — same pattern as
TopicResources for TopicListener. MetadataStore has no
unregisterListener, so per-session direct registration would leak listener
references for the broker's lifetime; the registry-and-fan-out approach
avoids both the leak and the linear per-event dispatch tax.

Cross-topic load balancing — deferred

Today: every multi-topic consumer in (namespace, filter, subscriptionName)
will subscribe to all matching topics. Each topic's per-topic
SubscriptionCoordinator independently picks one consumer per segment — for
StreamConsumer (Exclusive) the assignment is randomized across consumers, in
expectation roughly balanced but not deterministic.

Future: namespace-level MultiTopicSubscriptionCoordinator per
(namespace, subscriptionName) that:

• Persists Map<TopicName, ConsumerSession> for the group.
• On consumer attach: assigns a slice of the matching topic set.
• On topic Added/Removed: rebalances.
• Server-side filters the watcher's emitted set to the consumer's slice.

When that lands it'll add a consumer_name (or similar identity) field to
CommandWatchScalableTopics. Not added speculatively in this PR.

Authz

CommandWatchScalableTopics: requires namespace-level READ
(NamespaceOperation.GET_TOPICS, same as listScalableTopics). Per-topic
subscribe that follows uses normal topic-level authz.

Test plan

• ScalableTopicsWatcherSessionHashTest — broker session hash branches
  (no hash → snapshot; matching → silent; differing → snapshot). Mocked
  ServerCnx + LocalMemoryMetadataStore.
• V5ScalableTopicsWatcherTest — end-to-end through a real shared cluster:
  create / delete events flow as Diffs; AND property filters narrow the set
  correctly; pre-existing topics surface in the initial Snapshot.

Matching PR(s) in forked repositories

• area/broker
• area/client

[lhotari]

Note: this is analysis by Claude Code, reviewed by me before posting.

A few points worth checking before this merges.

1. Metadata listener leak per ScalableTopicsWatcherSession (ScalableTopicsWatcherSession.start():305, close():466-472)

start() calls resources.getStore().registerListener(this::onNotification) per session, and close() only flips a closed flag — MetadataStore exposes no unregisterListener. So every closed session leaves a stale Consumer<Notification> registered, and every metadata notification fans out to all stale listeners over the broker's lifetime (each short-circuits on closed.get(), but the dispatch cost still scales linearly with total sessions ever opened). Not just memory growth — a real per-event throughput tax for long-running brokers serving many namespace watches.

Suggested fix (primary) — mirror what TopicResources does for TopicListService:

• Have ScalableTopicResources register one handleNotification listener at construction time (TopicResources.java:52).
• Maintain a Map<ScalableTopicNamespaceListener, NamespaceName> (or similar) inside ScalableTopicResources, with register…Listener(...) / deregister…Listener(...) methods (TopicResources.java:136-142).
• ScalableTopicsWatcherSession.start() calls register…, close() calls deregister…. The single fan-out filters by the watcher's namespace base path.

This is the established pattern in the codebase, and it removes both the leak and the linear dispatch cost without needing a MetadataStore API change.

Alternative — #24256 adds registerCancellableListener returning a handle that callers cancel on shutdown. With that landed, ScalableTopicsWatcherSession could store the handle and cancel it in close(). Works, but keeps one metadata-store listener per session; the fan-out approach above is cheaper at runtime and consistent with TopicListService.

2. `consumer_name` described in PR body but absent from the proto (PulsarApi.proto:1558-1569)

The PR description's "Wire protocol" section states CommandWatchScalableTopics carries consumer_name, and the "Cross-topic load balancing — deferred" section says "`consumer_name` is part of `CommandWatchScalableTopics` now, so the future coordinator has the identity it needs." The proto in this PR has only watch_id, namespace, property_filters, current_hash. Either add the field now (cheap — optional string consumer_name = 5;) or remove the design language so we don't need another wire bump for the future coordinator.

3. Reconnect backoff never reset on hash-matched reconnect (ScalableTopicsWatcher.java:1037)

reconnectBackoff.reset() runs only inside onSnapshot. The hash-skip optimisation is the common short-blip happy path — the broker emits no Snapshot, so the backoff stays at its last value across successive successful reconnects. The next disconnect then waits much longer than expected. Suggest resetting on successful reconnect (after the watch frame's write completes, or on any successful broker→client traffic).

[lhotari]

LGTM, please check the local Claude Code review findings to see if they are real.

[merlimat]

All good points. Fixed