[improve][broker] PIP-473 P5.4: v4/v5 transaction coordinator coexistence + enable v5 by default

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -3798,13 +3798,14 @@ public double getLoadBalancerBandwidthOutResourceWeight() {
     @FieldContext(
             category = CATEGORY_TRANSACTION,
             doc = "Enable the metadata-driven transaction coordinator used by scalable topics."
-                    + " When true, wire commands (NEW_TXN / END_TXN / etc.) are served by the"
-                    + " metadata-store-backed coordinator instead of the legacy"
-                    + " TransactionMetadataStoreService. Requires transactionCoordinatorEnabled"
-                    + " = true, and must be enabled together with the scalable-topic transaction"
-                    + " buffer and pending-ack store providers."
+                    + " When true, transaction wire commands flagged as scalable (sent by v5 SDK"
+                    + " clients) are served by the metadata-store-backed coordinator, while legacy"
+                    + " (v4) clients continue to be served by TransactionMetadataStoreService — the"
+                    + " two coexist on the same cluster. Requires transactionCoordinatorEnabled"
+                    + " = true. Enabled by default together with the dispatching transaction buffer"
+                    + " and pending-ack store providers."
     )
-    private boolean transactionCoordinatorScalableTopicsEnabled = false;
+    private boolean transactionCoordinatorScalableTopicsEnabled = true;
 
     @FieldContext(
             category = CATEGORY_TRANSACTION,
@@ -3859,21 +3860,26 @@ public double getLoadBalancerBandwidthOutResourceWeight() {
 
     @FieldContext(
             category = CATEGORY_TRANSACTION,
-            doc = "Class name for transaction buffer provider. Default routes segment:// topics to the"
-                    + " legacy TopicTransactionBuffer. Set this to"
-                    + " org.apache.pulsar.broker.transaction.buffer.impl.DispatchingTransactionBufferProvider"
-                    + " once the v5 transaction coordinator (PIP-473 P5) is enabled to opt segment topics"
-                    + " into MetadataTransactionBuffer."
+            doc = "Class name for transaction buffer provider. The default DispatchingTransactionBufferProvider"
+                    + " routes segment:// topics to the metadata-driven MetadataTransactionBuffer (PIP-473)"
+                    + " and persistent:// / topic:// topics to the legacy TopicTransactionBuffer. Set this to"
+                    + " org.apache.pulsar.broker.transaction.buffer.impl.TopicTransactionBufferProvider to"
+                    + " force the legacy buffer for all topics."
     )
     private String transactionBufferProviderClassName =
-            "org.apache.pulsar.broker.transaction.buffer.impl.TopicTransactionBufferProvider";
+            "org.apache.pulsar.broker.transaction.buffer.impl.DispatchingTransactionBufferProvider";
 
     @FieldContext(
             category = CATEGORY_TRANSACTION,
-            doc = "Class name for transaction pending ack store provider"
+            doc = "Class name for transaction pending ack store provider. The default"
+                    + " DispatchingTransactionPendingAckStoreProvider routes subscriptions on segment:// topics"
+                    + " to the metadata-driven MetadataPendingAckStore (PIP-473) and others to the legacy"
+                    + " MLPendingAckStore. Set this to"
+                    + " org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStoreProvider to force"
+                    + " the legacy store for all subscriptions."
     )
     private String transactionPendingAckStoreProviderClassName =
-            "org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStoreProvider";
+            "org.apache.pulsar.broker.transaction.pendingack.impl.DispatchingTransactionPendingAckStoreProvider";
 
     @FieldContext(
             category = CATEGORY_TRANSACTION,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -3446,7 +3446,12 @@ protected void handleTcClientConnectRequest(CommandTcClientConnectRequest comman
             return;
         }
 
-        if (service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
+        if (command.isScalable()) {
+            if (!isScalableTcAvailable()) {
+                commandSender.sendTcClientConnectResponse(requestId, ServerError.NotAllowedError,
+                        "Scalable-topics transaction coordinator is not enabled on this broker");
+                return;
+            }
             service.pulsar().getTransactionCoordinatorV5().handleClientConnect(tcId)
                     .whenComplete((__, e) -> {
                         if (e == null) {
@@ -3492,6 +3497,16 @@ private boolean checkTransactionEnableAndSendError(long requestId) {
             return true;
         }
     }
+    /**
+     * @return true if the scalable-topics (PIP-473) transaction coordinator is enabled and ready on
+     *     this broker. Transaction commands carrying {@code scalable=true} route to it; commands
+     *     without the flag always go to the legacy coordinator, so v4 and v5 clients coexist.
+     */
+    private boolean isScalableTcAvailable() {
+        return service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()
+                && service.getPulsar().getTransactionCoordinatorV5() != null;
+    }
+
     private Throwable handleTxnException(Throwable ex, String op, long requestId) {
         Throwable cause = FutureUtil.unwrapCompletionException(ex);
         if (cause instanceof CoordinatorException.CoordinatorNotFoundException) {
@@ -3527,7 +3542,12 @@ protected void handleNewTxn(CommandNewTxn command) {
             return;
         }
 
-        if (service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
+        if (command.isScalable()) {
+            if (!isScalableTcAvailable()) {
+                commandSender.sendNewTxnErrorResponse(requestId, tcId.getId(), ServerError.NotAllowedError,
+                        "Scalable-topics transaction coordinator is not enabled on this broker");
+                return;
+            }
             final String v5Owner = getPrincipal();
             service.pulsar().getTransactionCoordinatorV5()
                     .newTransaction(tcId, command.getTxnTtlSeconds() * 1000L, v5Owner)
@@ -3594,11 +3614,17 @@ protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) {
             return;
         }
 
-        if (service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
+        if (command.isScalable()) {
+            if (!isScalableTcAvailable()) {
+                writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId, txnID.getLeastSigBits(),
+                        txnID.getMostSigBits(), ServerError.NotAllowedError,
+                        "Scalable-topics transaction coordinator is not enabled on this broker"));
+                return;
+            }
             // v5: TC doesn't need pre-registration — participants advertise themselves by writing
             // /txn/op records when they actually apply ops. Still verify ownership before acking,
             // matching the legacy authorization surface.
-            verifyTxnOwnership(txnID)
+            verifyTxnOwnership(txnID, true)
                     .thenCompose(isOwner -> isOwner ? CompletableFuture.<Void>completedFuture(null)
                             : failedFutureTxnNotOwned(txnID))
                     .whenComplete((v, ex) -> {
@@ -3618,7 +3644,7 @@ protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) {
 
         TransactionMetadataStoreService transactionMetadataStoreService =
                 service.pulsar().getTransactionMetadataStoreService();
-        verifyTxnOwnership(txnID)
+        verifyTxnOwnership(txnID, false)
                 .thenCompose(isOwner -> {
                     if (!isOwner) {
                         return failedFutureTxnNotOwned(txnID);
@@ -3676,8 +3702,13 @@ protected void handleEndTxn(CommandEndTxn command) {
             return;
         }
 
-        if (service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
-            verifyTxnOwnership(txnID)
+        if (command.isScalable()) {
+            if (!isScalableTcAvailable()) {
+                commandSender.sendEndTxnErrorResponse(requestId, txnID, ServerError.NotAllowedError,
+                        "Scalable-topics transaction coordinator is not enabled on this broker");
+                return;
+            }
+            verifyTxnOwnership(txnID, true)
                     .thenCompose(isOwner -> {
                         if (!isOwner) {
                             return failedFutureTxnNotOwned(txnID);
@@ -3700,7 +3731,7 @@ protected void handleEndTxn(CommandEndTxn command) {
         TransactionMetadataStoreService transactionMetadataStoreService =
                 service.pulsar().getTransactionMetadataStoreService();
 
-        verifyTxnOwnership(txnID)
+        verifyTxnOwnership(txnID, false)
                 .thenCompose(isOwner -> {
                     if (!isOwner) {
                         return failedFutureTxnNotOwned(txnID);
@@ -3739,14 +3770,13 @@ private CompletableFuture<Boolean> isSuperUser() {
         }
     }
 
-    private CompletableFuture<Boolean> verifyTxnOwnership(TxnID txnID) {
+    private CompletableFuture<Boolean> verifyTxnOwnership(TxnID txnID, boolean scalable) {
         assert ctx.executor().inEventLoop();
-        CompletableFuture<Boolean> ownerCheck =
-                service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()
-                        ? service.pulsar().getTransactionCoordinatorV5()
-                                .verifyTxnOwnership(txnID, getPrincipal())
-                        : service.pulsar().getTransactionMetadataStoreService()
-                                .verifyTxnOwnership(txnID, getPrincipal());
+        CompletableFuture<Boolean> ownerCheck = scalable
+                ? service.pulsar().getTransactionCoordinatorV5()
+                        .verifyTxnOwnership(txnID, getPrincipal())
+                : service.pulsar().getTransactionMetadataStoreService()
+                        .verifyTxnOwnership(txnID, getPrincipal());
         return ownerCheck
                 .thenComposeAsync(isOwner -> {
                     if (isOwner) {
@@ -4016,11 +4046,17 @@ protected void handleAddSubscriptionToTxn(CommandAddSubscriptionToTxn command) {
             return;
         }
 
-        if (service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
+        if (command.isScalable()) {
+            if (!isScalableTcAvailable()) {
+                writeAndFlush(Commands.newAddSubscriptionToTxnResponse(requestId, txnID.getLeastSigBits(),
+                        txnID.getMostSigBits(), ServerError.NotAllowedError,
+                        "Scalable-topics transaction coordinator is not enabled on this broker"));
+                return;
+            }
             // v5: TC doesn't need pre-registration — participants advertise themselves by writing
             // /txn/op records when they actually apply ops. Still verify ownership before acking,
             // matching the legacy authorization surface.
-            verifyTxnOwnership(txnID)
+            verifyTxnOwnership(txnID, true)
                     .thenCompose(isOwner -> isOwner ? CompletableFuture.<Void>completedFuture(null)
                             : failedFutureTxnNotOwned(txnID))
                     .whenComplete((v, ex) -> {
@@ -4041,7 +4077,7 @@ protected void handleAddSubscriptionToTxn(CommandAddSubscriptionToTxn command) {
         TransactionMetadataStoreService transactionMetadataStoreService =
                 service.pulsar().getTransactionMetadataStoreService();
 
-        verifyTxnOwnership(txnID)
+        verifyTxnOwnership(txnID, false)
                 .thenCompose(isOwner -> {
                     if (!isOwner) {
                         return failedFutureTxnNotOwned(txnID);

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBuffer.java

@@ -119,9 +119,6 @@ public class MetadataTransactionBuffer implements TransactionBuffer {
     /** Version of the durable watermark record; -1 if it doesn't exist yet. */
     private long watermarkVersion = -1L;
 
-    /** Latest dispatched position from non-txn publishes — the natural ceiling when no opens pin. */
-    private Position lastDispatchable;
-
     /** Current maxReadPosition; never moves above the watermark while recovery-discovered opens exist. */
     private Position maxReadPosition;
 
@@ -141,7 +138,6 @@ public MetadataTransactionBuffer(PersistentTopic topic, TxnMetadataStore txnStor
         this.segmentName = topic.getName();
         this.maxReadPositionCallBack = topic.getMaxReadPositionCallBack();
         this.maxReadPosition = ledger.getLastConfirmedEntry();
-        this.lastDispatchable = this.maxReadPosition;
         recover();
     }
 
@@ -544,18 +540,28 @@ private void recomputeMaxReadPositionLocked() {
             next = watermarkPos != null ? watermarkPos : maxReadPosition;
         } else {
             Position min = null;
+            boolean anyOpen = false;
             for (TxnEntry e : txns.values()) {
-                if (e.state == TxnState.OPEN && e.firstPosition != null) {
-                    if (min == null || e.firstPosition.compareTo(min) < 0) {
+                if (e.state == TxnState.OPEN) {
+                    anyOpen = true;
+                    if (e.firstPosition != null
+                            && (min == null || e.firstPosition.compareTo(min) < 0)) {
                         min = e.firstPosition;
                     }
                 }
             }
             if (min != null) {
+                // Pin just below the lowest open txn's first write.
                 next = ledger.getPreviousPosition(min);
+            } else if (anyOpen) {
+                // Open txn(s) whose first write isn't tracked yet (an append is in flight between
+                // the /txn/op record and the ledger entry): hold the current ceiling rather than
+                // risk exposing the in-flight entry.
+                next = maxReadPosition;
             } else {
-                // No open txns pinning anything: free to advance to last-dispatched.
-                next = lastDispatchable;
+                // No open txns: every written entry is resolved — committed data is visible and
+                // aborted data is filtered by isTxnAborted — so advance to the last written entry.
+                next = ledger.getLastConfirmedEntry();
             }
         }
         Position prev = maxReadPosition;
@@ -593,7 +599,6 @@ public void syncMaxReadPositionForNormalPublish(Position position, boolean isMar
         }
         topic.updateLastDispatchablePosition(position);
         synchronized (lock) {
-            lastDispatchable = position;
             recomputeMaxReadPositionLocked();
             // Persist the new watermark if it advanced as a result of the non-txn append.
             stateTail = stateTail.thenCompose(__ -> persistWatermarkIfAdvanced())

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStore.java

@@ -30,6 +30,7 @@
 import lombok.CustomLog;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.pulsar.broker.service.BrokerServiceException;
 import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.transaction.metadata.TxnIds;
 import org.apache.pulsar.broker.transaction.metadata.TxnMetadataStore;
@@ -165,7 +166,21 @@ public void onCompleted() {
                 return;
             }
             recoveryFuture.complete(null);
-            pendingAckHandle.changeToReadyState();
+            // Mirror the legacy MLPendingAckStore completion: flip the handle to Ready and
+            // complete the handle future — PersistentSubscription.addConsumer blocks on that
+            // future, so skipping it hangs every subscribe on a segment topic — then drain any
+            // ack requests queued during recovery. Run on the pinned executor so the
+            // state-machine transition and the cache drain stay single-threaded.
+            executorService.execute(() -> {
+                if (pendingAckHandle.changeToReadyState()) {
+                    pendingAckHandle.completeHandleFuture();
+                } else {
+                    pendingAckHandle.exceptionHandleFuture(
+                            new BrokerServiceException.ServiceUnitNotReadyException(
+                                    "Failed to change PendingAckHandle state to Ready"));
+                }
+                pendingAckHandle.handleCacheRequest();
+            });
             // Drain any events that fired during recovery.
             triggerReconcile();
         });

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckPersistentTest.java

@@ -166,8 +166,11 @@ public void testBuildConsumerEncounterPendingAckInitFailure(Exception retryableE
         when(mockProvider.newPendingAckStore(any()))
                 // First, the method newPendingAckStore will fail with a retryable exception.
                 .thenReturn(FutureUtil.failedFuture(new ManagedLedgerException("mock fail new store")))
-                // Then, the method will be executed successfully.
-                .thenCallRealMethod();
+                // Then, the method will be executed successfully. Delegate to the real provider
+                // rather than thenCallRealMethod(): the configured provider is now the dispatching
+                // provider, and a Mockito mock of it has null delegate fields, so calling its real
+                // method would NPE. The original real provider behaves identically for this topic.
+                .thenAnswer(invocation -> pendingAckStoreProvider.newPendingAckStore(invocation.getArgument(0)));
         transactionPendingAckStoreProviderField.set(pulsarServiceList.get(0), mockProvider);
         Consumer<byte[]> consumer3 = pulsarClient.newConsumer()
                 .subscriptionName("subName3")

pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5.java

@@ -43,6 +43,10 @@ final class PulsarClientBuilderV5 implements PulsarClientBuilder {
 
     PulsarClientBuilderV5() {
         conf.setStatsIntervalSeconds(0);
+        // v5 SDK transactions use the metadata-store (PIP-473) coordinator. This internal flag
+        // routes the underlying v4 TC client to it, keeping v5 transactions independent from any
+        // v4 SDK clients (which use the legacy coordinator) on the same cluster.
+        conf.setScalableTransactions(true);
     }
 
     @Override