[improve][broker] PIP-473 P5.1: metadata-driven transaction coordinator (NEW_TXN / END_TXN)#25863
Merged
Merged
Conversation
… NEW_TXN / END_TXN Adds the v5 transaction coordinator behind the new `transactionCoordinatorV5Enabled` flag (default off). When the flag is on, `ServerCnx` routes `TC_CLIENT_CONNECT`, `NEW_TXN`, `END_TXN`, `ADD_PARTITION_TO_TXN`, and `ADD_SUBSCRIPTION_TO_TXN` to `TransactionCoordinatorV5` instead of the legacy `TransactionMetadataStoreService`. The TC writes the txn header at `/txn/id/<tcId>_<seq>`, assigns `seq` from a per-tcId monotonic counter (`/txn/tc-seq/<tcId>`) so txnIds are never reused, and at end-txn enumerates the txn's participants via the new `idx:ops-by-txn` secondary index on `/txn/op`, publishing one segment-event per affected segment and one subscription-event per affected (segment, subscription) pair. `ADD_PARTITION_TO_TXN` / `ADD_SUBSCRIPTION_TO_TXN` are no-ops per PIP-473 — participants advertise themselves by writing `/txn/op` records when they apply ops, so the pre-registration step is unnecessary. Leader election reuses the existing `transaction_coordinator_assign` topic ownership check (same surface the legacy TC uses); a pure metadata-store election lands in P5.3.
Addresses review feedback: - Rename `transactionCoordinatorV5Enabled` → `transactionCoordinatorScalableTopicsEnabled` and drop PIP / phase references from the user-facing config doc. - Import `TransactionCoordinatorV5` in `PulsarService` instead of using the fully qualified name inline.
lhotari
reviewed
May 25, 2026
…y fan-out - Security: the v5 ServerCnx paths bypassed the ownership check the legacy handlers run via verifyTxnOwnership. Persist the opening principal in TxnHeader.owner (NEW_TXN), add TransactionCoordinatorV5.verifyTxnOwnership (null owner ⟹ allowed, mirroring the legacy semantics), route ServerCnx.verifyTxnOwnership to the v5 TC when the flag is on, and gate END_TXN / ADD_PARTITION / ADD_SUBSCRIPTION on it (superuser fallback unchanged). - endTransaction idempotent retry now re-drives the fan-out instead of short-circuiting: if a prior attempt CAS'd the header but failed before publishing, a terminal header gives the reconcile path nothing to act on, so the retry must re-publish. Safe — participants key on (txnId, decision) and the decision can't change once terminal. - Replace the hand-rolled ConcurrentHashSet helper with ConcurrentHashMap.newKeySet(), and the "\0"-packed (segment, subscription) string with an AckParticipant record. Tests: owner persistence + verifyTxnOwnership (match / null-allowed / unknown), idempotent-retry re-publish. Existing P2/P3/P4 TxnHeader constructions updated for the new owner field.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds the PIP-473 metadata-driven v5 transaction coordinator — the broker-side service that serves
NEW_TXN/END_TXNagainst the metadata-store transaction layout (P2'sTxnMetadataStore) instead of the legacyTransactionMetadataStoreService.Gated behind a new
transactionCoordinatorScalableTopicsEnabledflag (default off). When on,ServerCnxroutes the TC wire commands toTransactionCoordinatorV5; when off, behavior is unchanged. The flag is meant to be flipped together with the scalable-topic transaction buffer and pending-ack store providers (a later phase).Leader election
Per-partition coordinator, same as the legacy TC: a broker runs the v5 TC for partition
Niff it owns partitionNoftransaction_coordinator_assign.handleClientConnectmirrors the legacy ownership check, so the client-side discovery surface is unchanged. (A pure metadata-store election is a later phase; reusing the assign-topic keeps this PR's scope tight.)Wire commands
TC_CLIENT_CONNECTNEW_TXN/txn/id/<tcId>_<seq>header inOPENEND_TXNCOMMITTED/ABORTED, fan out participant eventsADD_PARTITION_TO_TXN,ADD_SUBSCRIPTION_TO_TXNADD_PARTITION/ADD_SUBSCRIPTIONare no-ops per the PIP: v5 participants advertise themselves by writing/txn/oprecords when they apply ops, so the pre-registration step is unnecessary.txnId generation — no reuse
leastSigBitsis drawn from a per-tcId monotonic counter at/txn/tc-seq/<tcId>(TcSequence, CAS-incremented with retry onBadVersionException). Monotonic-per-tcId ⟹ txnIds are never reused — the participant-side aborted set is keyed by txnId, and reuse would corrupt it.endTransaction fan-out
END_TXNCAS-updates the header to the terminal state, then enumerates the txn's participants via the newidx:ops-by-txnsecondary index on/txn/op(so it doesn't scan the whole namespace), and publishes one segment-event per affected segment + one subscription-event per affected(segment, subscription)pair. Idempotent on retry — a header already in the requested terminal state short-circuits.Test plan
pulsar-broker:test --tests TransactionCoordinatorV5Test— 8 cases: sequential txnId per tcId, commit/abort fan-out, idempotent retry, mismatched-action failure, unknown-txn failure, add* no-ops, per-tcId scoping.pulsar-broker:test --tests TxnMetadataStoreTest/MetadataTransactionBufferTest/MetadataPendingAckStoreTest— green after theidx:ops-by-txnindex addition.Deferred / follow-ups
newTxn/endTxnonly.