[improve][broker] PIP-473 P5.3: metadata-store TC leader election + assignment watch

[merlimat]

Motivation

The scalable-topics transaction coordinator (PIP-473) so far reused the
transaction_coordinator_assign partitioned topic for two jobs: per-partition leader election and client discovery. Both route through the topic / namespace / bundle / load-balancer machinery, so TC coordination liveness is entangled with that machinery — even though the TC keeps no state in the assign-topic (its ledgers are empty; it's purely an ownership token).

This PR makes the scalable-topics TC's leader election depend on the metadata store only, and replaces the topic-lookup-based discovery with a dedicated assignment watch. The coordinator now rests directly on the metadata store it already hard-depends on for every header read/write, instead of on a layer above it.

Note: sharding is unchanged — placement is still client-side round-robin and the TC partition is encoded in TxnID.mostSigBits. Only how a broker becomes leader for partition N and how the client finds that broker change.

Modifications

Broker election

• Per-partition LeaderElection<TcLeader> over /txn/tc/leader/<N> via the existing CoordinationService.
• New config transactionCoordinatorScalableTopicsParallelism (default 16) — the degree of coordinator parallelism; replaces the assign-topic partition count for the v5 path.
• TransactionCoordinatorV5.isLeaderFor(int) now gates both client-connect and the timeout/GC sweeps (was the assign-topic ownership check).

Discovery (new wire commands)

• CommandWatchTcAssignments / CommandWatchTcAssignmentsUpdate / CommandWatchTcAssignmentsClose + TcAssignment / TcAssignmentsSnapshot.
• The client opens one watch; the broker replies with the full partition → leader map and re-pushes the full snapshot on every leadership change. No point lookup, no diff/hash — the map is bounded (parallelism, ~16) and changes rarely, so always sending the whole snapshot is simpler and removes a class of apply-ordering/drift bugs.
• New FeatureFlags.supports_tc_metadata_discovery.

Client

• TcDiscovery strategy: WatchTcAssignmentsDiscovery (new path) when the broker advertises the feature flag, else AssignTopicTcDiscovery (the existing flow — still the v4 path).
• TransactionMetaStoreHandler can connect directly to a coordinator's elected leader broker and retarget when a snapshot moves leadership.

Backward compatibility

Purely additive on the wire. The assign-topic remains the v4 / fallback discovery surface during the deprecation window:

• new client + new broker → assignment watch;
• old client + new broker → assign-topic lookup (broker still owns the bundle; TC_CLIENT_CONNECT still works);
• new client + old broker → feature flag absent → falls back to assign-topic.

Defaults are unchanged — the scalable-topics TC stays off; the default flip lands in P5.4.

Tests

• TransactionCoordinatorV5Test — election + leadership coverage (21 cases).
• CommandsTcAssignmentsTest — wire round-trip (5 cases).
• TcMetadataDiscoveryTest — multi-broker docker integration: transaction lifecycle over the assignment watch across all coordinator partitions, and survival of a leader-broker failure (re-election → watch refresh → handler retarget). Wired into the TRANSACTION CI group.
• Checkstyle clean across broker / client / common / integration.

Scope note: the integration tests exercise the transaction lifecycle (newTransaction / commit / abort) over the discovered connections — the full surface the new client discovery path drives. A data-in-transaction e2e (produce/ack inside a txn) additionally needs the scalable-topic buffer + pending-ack providers and segment:// topics, which land with P5.4 / P6.

[lhotari]

the TC partition is encoded in TxnID.mostSigBits

Is is possible to modify transactionCoordinatorScalableTopicsParallelism afterwards? Since aborted transactions will stay around as long as the messages participating in the transaction are retained, does that cause additional constraints?

[lhotari]

These are the findings of a local Claude Code review. Please check whether these need to be addressed.

1. [BUG] Follower-broker snapshot push races the leader-value cache; no re-push → client stuck on a dead leader indefinitely

TransactionCoordinatorV5.buildAssignmentsSnapshot + onElectionStateChange

Trace for a 3+ broker cluster (A leads partition N; B and C follow; the client's watch is on C):

• A dies → Deleted notification. MetadataCacheImpl.accept invalidates C's cache entry (MetadataCacheImpl.java:378-379). LeaderElectionImpl.handlePathNotification silently sets NoLeader and re-elects.
• C loses to B → changeState(Following) fires C's listener synchronously inside handleExistingLeaderValue (LeaderElectionImpl.java:163) — before the cache.refresh(path) / cache.get(path) continuation in elect() (LeaderElectionImpl.java:115-121) repopulates the cache (that's a metadata-store RTT away).
• The ServerCnx listener task (ctx.executor().execute(...), sub-millisecond) builds the snapshot via getLeaderValueIfPresent() = cache.getIfCached(path) → empty → partition N is omitted from the pushed snapshot. Note that MetadataCacheImpl.refresh() is computeIfPresent-only (MetadataCacheImpl.java:358-361), so the Created notification for B's node does not repopulate an invalidated entry either.
• There is no subsequent trigger: the cache repopulating fires no TC listener, the broker has no periodic/delayed re-push, and the client never re-polls. The client's handler for partition N keeps reconnect-looping against dead broker A "until the next snapshot" — which only arrives if some other leadership change happens later.

The PR description's claim that the client parks a mid-election partition "until a later snapshot fills it in" is therefore not upheld — the later snapshot may never come.

The 2-broker integration test structurally masks this (PulsarClusterSpec.numBrokers = 2): the survivor transitions Following→Leading for every moved partition, and in the Leading path tryToBecomeLeader populates the cache before firing the listener (LeaderElectionImpl.java:194-208), so its snapshots are always correct.

Suggested fixes: build the snapshot from getLeaderValue() (async, loads on miss) instead of getLeaderValueIfPresent(), and/or debounce-re-push a short delay after any change, and/or re-push when a snapshot was incomplete.

2. [BUG] Two paths where the client watch dies silently with no reconnect

WatchTcAssignmentsDiscovery

• attach(): on a reconnect (initial snapshot already received), if the newly-connected broker doesn't advertise supports_tc_metadata_discovery (e.g. per-broker config drift on transactionCoordinatorScalableTopicsEnabled), initialSnapshotFuture.completeExceptionally(...) is a no-op on the already-completed future and the method returns — no scheduleReconnect(). Discovery is permanently frozen on the last snapshot.
• onError(): post-initial errors only log. ClientCnx.handleCommandWatchTcAssignmentsUpdate has already removed the session on error, so no connectionClosed() will ever fire for it either. A broker answering NotAllowedError / ServiceNotReady after a reconnect kills the watch forever.

Both leave all handlers pinned to stale leaders with no recovery. Combined with finding 1, eventual consistency of the assignment map is not guaranteed.

3. [BUG] Initial watch open is one-shot — a transient failure hard-fails transaction client startup

TransactionCoordinatorClientImpl.selectDiscovery + WatchTcAssignmentsDiscovery.start

The probe (getConnectionToServiceUrl()) and the watch (getAnyBrokerProxyConnection()) can land on different brokers. If the watch broker rejects the open — e.g. ServiceNotReady while its TC is still initializing, or the feature flag is off there due to config drift — attach() / onError() fail initialSnapshotFuture → startAsync() fails outright. There is no retry for retryable errors, so transaction-enabled client startup can fail nondeterministically. Consider retrying the watch open (with the existing reconnectBackoff) for retryable failures instead of failing startup.

4. [QUALITY] new URI(null) NPE for a non-TLS client when a leader advertises only a TLS URL

TransactionMetaStoreHandler ctor / retargetLeader

HandlerState.setRedirectedClusterURI picks serviceUrl when !isUseTls() (HandlerState.java:56-59); TcAssignment.broker_service_url is optional and the snapshot delivers null for TLS-only brokers. new URI(null) throws NPE, not URISyntaxException — so the constructor's catch doesn't wrap it and retargetLeader's catch doesn't swallow it. It propagates out of handlers.compute inside onSnapshot on the netty read thread → channel teardown → reconnect → repeat. A misconfigured client, but the failure mode should be a clear error, not an exception loop.

5. [QUALITY] transactionCoordinatorScalableTopicsParallelism has no validation or cross-broker consistency enforcement

ServiceConfiguration.java

No minValue = 1 on the @FieldContext (other int configs set it; 0/negative silently yields a coordinator that leads nothing and nextHandler() == null). More importantly, the doc says "Fixed at cluster bring-up," but nothing enforces agreement: brokers with different values run different election sets, and the snapshot's parallelism flip-flops depending on which broker serves the watch. Consider persisting the value in the metadata store on first start and failing/warning on mismatch.

6. [QUALITY] The integration test can't prove the new path is exercised

TcMetadataDiscoveryTest

If selectDiscovery's probe transiently fails, it silently falls back to AssignTopicTcDiscovery — and both tests still pass, since the assign topic is initialized with the same partition count (TC_PARALLELISM). Consider asserting the selected discovery type (e.g. via the "Transaction coordinator discovery selected" log line or a test hook); otherwise a regression that breaks the watch path entirely could go green in CI.

7. [QUALITY] Minor notes

• AssignTopicTcDiscovery.handlers() reads the handlers field twice (handlers == null ? List.of() : List.of(handlers)) — a concurrent close() between the two reads NPEs. Test-only surface, trivial to fix with a local variable.
• On broker startup with 16 partitions, each elect completion fires every watcher listener → a connecting client may receive a burst of ~16 full snapshots. Bounded and harmless, but a debounce (which finding 1 wants anyway) would also fix this.

[lhotari]

LGTM, however please check the local Claude Code review findings about the possible bugs.

[merlimat]

Thanks for the very thorough review — went through every finding against the code; all were legitimate. Addressed in 6acca7f.

On the design question (can parallelism change after bring-up?): No, and your instinct about aborted transactions is exactly the additional constraint. TxnID.mostSigBits encodes the coordinator, so changing parallelism strands existing transaction ids. And because an aborted transaction's records are retained as long as its participating messages are, the value can only ever be reduced once every transaction created under the previous value has been fully cleaned up. I've documented this on the config and now enforce cross-broker agreement (see #5 below).

Findings:

1. [BUG] follower-snapshot race — confirmed. buildAssignmentsSnapshot used the cache-only getLeaderValueIfPresent(); switched to the async getLeaderValue() (loads from the store on a cache miss), so a freshly-Following broker no longer omits the partition. Added a short re-push from ServerCnx when a snapshot is incomplete or fails to build, so the client converges without relying on a later leadership change. Note: the deepest 3+-broker scenario isn't reproducible by the 2-broker integration test, so I'm relying on the mechanism fix + re-push rather than an e2e proof — flagging that honestly.

2. [BUG] watch dies silently — confirmed both paths. attach() on reconnect without the flag, and post-initial onError() (you're right that ClientCnx already removed the session so no connectionClosed() fires) now both scheduleReconnect().

3. [BUG] one-shot initial open — confirmed. The initial open now retries retryable failures up to the lookup deadline instead of failing startAsync().

4. [QUALITY] new URI(null) NPE — confirmed it's an NPE not URISyntaxException. HandlerState.setRedirectedClusterURI now throws a clear URISyntaxException for a blank selected URL, and onSnapshot isolates per-partition failures.

5. [QUALITY] parallelism validation/consistency — added minValue=1; the value is persisted at /txn/tc/parallelism on first start and every broker fails fast on mismatch. New unit test.

6. [QUALITY] test can't prove the new path — fixed; the integration test now asserts isUsingMetadataDiscovery() so a watch-path regression can't pass via the silent assign-topic fallback.

7. [minor] handlers() now reads the field into a local; the javascript client #1 re-push also bounds the startup snapshot burst.

All affected unit tests + checkstyle pass locally; the integration test compiles and is wired into the TRANSACTION suite.

[lhotari]

just wondering what happens when the target broker goes down? does the retry loop exit?

[lhotari]

@merlimat This PR caused a regression, filed as #25997. Could you please take a look? I think that it's a release blocker for 5.0.0-M1.