[improve][broker] PIP-473 P5.4: v4/v5 transaction coordinator coexistence + enable v5 by default#25945
Merged
Merged
Conversation
…d v5 coexist Enabling the scalable-topics (v5) transaction coordinator previously sent ALL clients' NEW_TXN/END_TXN/etc. to it via a global config check, which would break v4 transactions on persistent:// topics (the v5 TC notifies participants via metadata-store events, not the legacy RPC commit a v4 TopicTransactionBuffer waits for). Route per client instead so both coordinators serve their own clients on the same cluster. - Add an optional `scalable` flag to CommandTcClientConnectRequest, CommandNewTxn, CommandEndTxn, CommandAddPartitionToTxn, CommandAddSubscriptionToTxn (additive, defaults false) + Commands builder overloads. - ServerCnx routes each command to the v5 TC when scalable=true (rejecting it if the v5 TC isn't enabled), else to the legacy TC — replacing the global-config branch. verifyTxnOwnership is parameterized by the same flag. - The v5 client (metadata-store discovery: handlers built with a leader URI) sets scalable=true on every txn command; the v4/assign-topic path leaves it unset. No shared txnId namespace needed — each TC keeps its own state, and a client only ever talks to one TC kind, so a bare TxnID is never looked up across both. Tests: Commands round-trip of the flag on all five commands; integration test legacyTransactionStillWorksWhenScalableTcEnabled runs a v4 transaction on a persistent topic against a v5-enabled cluster. Verified locally against a built docker image — all three TcMetadataDiscoveryTest methods pass.
…actions by default Routes transaction coordination by client/SDK kind rather than broker capability, then flips the defaults so the scalable-topics (v5) coordinator is on out of the box while v4 transactions keep working — both SDKs coexist on one cluster. - ClientConfigurationData.scalableTransactions: internal flag (not on the public ClientBuilder). PulsarClientBuilderV5 sets it true; the v4 builder never does. - selectDiscovery() routes by that flag: v5 SDK -> metadata-store coordinator (assignment watch), v4 SDK -> legacy assign-topic coordinator. Previously it keyed off the broker's supports_tc_metadata_discovery feature flag, which once the default was enabled would re-route v4 clients to the v5 TC and break their transactions (the v5 TC notifies participants via metadata-store events the legacy buffer / pending-ack store don't consume). A v5 SDK on an http:// URL now fails clearly instead of silently downgrading. - Defaults flipped: transactionCoordinatorScalableTopicsEnabled=true, transactionBufferProviderClassName=DispatchingTransactionBufferProvider, transactionPendingAckStoreProviderClassName=DispatchingTransactionPendingAckStoreProvider. The dispatching providers route segment:// topics to the metadata-driven implementations and persistent:// to the legacy ones, so v4 topics are unaffected. Integration test (verified locally on a built docker image, all PASS): a v5 SDK client drives the lifecycle + survives a leader-broker kill via the metadata coordinator, while a v4 SDK client runs a transaction on a persistent:// topic via the legacy coordinator — concurrently, on the same v5-enabled cluster. Asserts the v4 client does NOT use metadata discovery.
… broker Motivation: The P5.4 default flip activates MetadataTransactionBuffer and MetadataPendingAckStore in a live broker for the first time (previously only exercised against mocks), exposing two latent bugs that broke v5 transactions on scalable topics, plus two perf-tool gaps on that path. Modifications: - MetadataPendingAckStore.replayAsync: on recovery, flip the handle to Ready, complete the pending-ack handle future, and drain queued ack requests (mirroring the legacy MLPendingAckStore). Without completing that future, PersistentSubscription.addConsumer blocked forever and every segment-topic subscribe timed out. - MetadataTransactionBuffer.recomputeMaxReadPositionLocked: once no open txns remain, advance the max read position to the last confirmed entry (it was pinned at the last non-transactional publish, so committed transactional data never became visible). Hold while an open txn's positions aren't tracked yet (in-flight append); drop the now-unused lastDispatchable field. - testclient: open the initial transaction with a bounded retry while the transaction-coordinator handler connects (avoids MetaStoreHandlerNotReadyException), and let PerformanceConsumer terminate on idle receives so an async commit landing after the last consumed message no longer hangs it. - Convert PerformanceTransactionTest and Oauth2PerformanceTransactionTest to scalable topics + the v5 SDK verification client; Performance Transaction gains --scalable. Assisted-by: Claude Code
…connecting
Motivation:
With the scalable-topics (v5) coordinator, the transaction-meta-store
handler resolves its partition leader asynchronously via the assignment
watch, so it sits in the Uninitialized state briefly after the client is
built. A first newTransaction in that window failed fast with
MetaStoreHandlerNotReadyException ("not connected"), which broke
V5ClientLifecycleTest.testNewTransactionAsyncReturnsOpenTransaction and
forced perf tools to retry.
Modifications:
- TransactionMetaStoreHandler.checkStateAndSendRequest: treat Uninitialized
like Connecting — leave the op queued in pendingRequests so it is retried
from connectionOpened once the handler is Ready, instead of failing fast.
The operation-timeout sweep still fails the op if the connection never
comes. Failed/Closing/Closed keep their existing fail-fast behavior, so
TransactionClientConnectTest's Closed-state assertion is unaffected.
Assisted-by: Claude Code
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.
Motivation
The metadata-driven (v5, PIP-473) transaction coordinator and the legacy (v4) coordinator must serve their own clients on the same cluster, and the v5 coordinator should be enabled by default. Before this change, the broker chose the coordinator from a global config flag, so enabling the v5 TC routed every client — including v4 SDK clients on
persistent://topics — to the v5 coordinator. That breaks v4 transactions: the v5 TC notifies participants via metadata-store events, which the legacy transaction buffer / pending-ack store don't consume.This PR routes coordination by client/SDK kind instead of broker capability, then flips the defaults so v5 is on out of the box while v4 keeps working.
Modifications
Routing by SDK kind
ClientConfigurationData.scalableTransactionsflag — not on the publicClientBuilder.PulsarClientBuilderV5sets ittrue; the v4 builder never does.TransactionCoordinatorClientImpl.selectDiscovery()routes by that flag: v5 SDK → metadata-store coordinator (assignment watch), v4 SDK → legacy assign-topic coordinator. A v5 SDK on anhttp://service URL now fails clearly rather than silently downgrading.scalableflag onCommandTcClientConnectRequest,CommandNewTxn,CommandEndTxn,CommandAddPartitionToTxn,CommandAddSubscriptionToTxn(additive, defaults false).ServerCnxroutes each command to the v5 TC whenscalable=true(rejecting it if the v5 TC isn't enabled), else to the legacy TC;verifyTxnOwnershipis parameterized by the same flag.No shared
TxnIDnamespace is needed: the two coordinators keep separate state, and a client only ever talks to one kind, so a bareTxnIDis never looked up across both.Default flip (P5.4)
transactionCoordinatorScalableTopicsEnabled→truetransactionBufferProviderClassName→DispatchingTransactionBufferProvidertransactionPendingAckStoreProviderClassName→DispatchingTransactionPendingAckStoreProviderThe dispatching providers route
segment://topics to the metadata-driven implementations andpersistent:///topic://to the legacy ones byTopicName.isSegment(), so v4 topics get the unchanged legacy buffer/ack-store.Verification
CommandsScalableTxnFlagTestround-trips thescalableflag on all five commands (present + default-false).TcMetadataDiscoveryTest, verified locally against a freshly built docker image — all pass): on a v5-enabled cluster, a v5 SDK client drives the transaction lifecycle and survives a leader-broker kill via the metadata coordinator, while a v4 SDK client runs a transaction on apersistent://topic via the legacy coordinator — concurrently. Asserts the v4 client does not use metadata-store discovery.Notes for reviewers
scalableflag); flagging for protocol review.