Apply changes to related classes.

dao-jun · May 14, 2024 · 3a745a6 · 3a745a6
1 parent 23d484a
commit 3a745a6
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 4 deletions.
diff --git a/...rc/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java b/...rc/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);
             commitFuture.complete(null);
         } catch (TransactionBufferException.TransactionNotFoundException
                 | TransactionBufferException.TransactionStatusException e) {
@@ -331,6 +332,7 @@ public CompletableFuture<Void> abortTxn(TxnID txnID, long lowWaterMark) {
             TxnBuffer txnBuffer = getTxnBufferOrThrowNotFoundException(txnID);
             txnBuffer.abort();
             buffers.remove(txnID, txnBuffer);
+            updateLastDispatchablePosition(null);
             abortFuture.complete(null);
         } catch (TransactionBufferException.TransactionNotFoundException
                 | TransactionBufferException.TransactionStatusException e) {
@@ -377,8 +379,11 @@ public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) {
 
     @Override
     public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
-        if (!isMarkerMessage && maxReadPositionCallBack != null) {
-            maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+        if (!isMarkerMessage) {
+            updateLastDispatchablePosition(position);
+            if (maxReadPositionCallBack != null) {
+                maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+            }
         }
     }
 
@@ -436,4 +441,11 @@ public long getCommittedTxnCount() {
                 .filter(txnBuffer -> txnBuffer.status.equals(TxnStatus.COMMITTED))
                 .count();
     }
+
+    // ThreadSafe
+    private void updateLastDispatchablePosition(Position position) {
+        if (topic instanceof PersistentTopic t) {
+            t.updateLastDispatchablePosition(position);
+        }
+    }
 }
diff --git a/.../main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java b/.../main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
@@ -99,8 +99,11 @@ public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) {
 
     @Override
     public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
-        if (!isMarkerMessage && maxReadPositionCallBack != null) {
-            maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+        if (!isMarkerMessage) {
+            updateLastDispatchablePosition(position);
+            if (maxReadPositionCallBack != null) {
+                maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+            }
         }
     }
 
@@ -148,4 +151,11 @@ public long getAbortedTxnCount() {
     public long getCommittedTxnCount() {
         return 0;
     }
+
+    // ThreadSafe
+    private void updateLastDispatchablePosition(Position position) {
+        if (topic instanceof PersistentTopic t) {
+            t.updateLastDispatchablePosition(position);
+        }
+    }
 }