[fix][broker] fix getMaxReadPosition in TransactionBufferDisable should return latest (apache#24898)

Co-authored-by: fanjianye <fanjianye@bigo.sg>
(cherry picked from commit b297f1f)

Author: 2 people

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -3731,6 +3731,12 @@ public CompletableFuture<Position> getLastDispatchablePosition() {
         if (lastDispatchablePosition != null) {
             return CompletableFuture.completedFuture(lastDispatchablePosition);
         }
+        PositionImpl lastPosition;
+        if (transactionBuffer instanceof TransactionBufferDisable) {
+            lastPosition = (PositionImpl) getLastPosition();
+        } else {
+            lastPosition = getMaxReadPosition();
+        }
         return ManagedLedgerImplUtils
                 .asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> {
                     MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer());
@@ -3743,7 +3749,7 @@ public CompletableFuture<Position> getLastDispatchablePosition() {
                         return !isTxnAborted(txnID, (PositionImpl) entry.getPosition());
                     }
                     return true;
-                }, getMaxReadPosition())
+                }, lastPosition)
                 .thenApply(position -> {
                     // Update lastDispatchablePosition to the given position
                     updateLastDispatchablePosition(position);

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

@@ -107,7 +107,7 @@ public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean i
 
     @Override
     public PositionImpl getMaxReadPosition() {
-        return (PositionImpl) topic.getLastPosition();
+        return PositionImpl.LATEST;
     }
 
     @Override