[fix][broker] Check the markDeletePosition and calculate the backlog (a…

…pache#22947) Signed-off-by: Zixuan Liu <[email protected]> (cherry picked from commit 82b8d98) (cherry picked from commit c0e1bff)
datastax · Jul 1, 2024 · 8edc5ab · 8edc5ab
1 parent 7d13506
commit 8edc5ab
Showing 1 changed file with 9 additions and 5 deletions.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -1112,6 +1112,13 @@ public long getEstimatedSizeSinceMarkDeletePosition() {
         return ledger.estimateBacklogFromPosition(markDeletePosition);
     }
 
+    private long getNumberOfEntriesInBacklog() {
+        if (markDeletePosition.compareTo(ledger.getLastPosition()) >= 0) {
+            return 0;
+        }
+        return getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
+    }
+
     @Override
     public long getNumberOfEntriesInBacklog(boolean isPrecise) {
         if (log.isDebugEnabled()) {
@@ -1120,16 +1127,13 @@ public long getNumberOfEntriesInBacklog(boolean isPrecise) {
                     messagesConsumedCounter, markDeletePosition, readPosition);
         }
         if (isPrecise) {
-            if (markDeletePosition.compareTo(ledger.getLastPosition()) >= 0) {
-                return 0;
-            }
-            return getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
+            return getNumberOfEntriesInBacklog();
         }
 
         long backlog = ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER.get(ledger) - messagesConsumedCounter;
         if (backlog < 0) {
             // In some case the counters get incorrect values, fall back to the precise backlog count
-            backlog = getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
+            backlog = getNumberOfEntriesInBacklog();
         }
 
         return backlog;