You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/06/20 18:46:47 UTC
(pulsar) 03/03: [fix][broker] Check the markDeletePosition and calculate the backlog (#22947)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c0e1bff5ace138d1830be19480da149725bb6142
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Jun 20 21:47:27 2024 +0800
[fix][broker] Check the markDeletePosition and calculate the backlog (#22947)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 82b8d98a488191d279612d5cf2b4846627863543)
---
.../apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 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
index 1292f15d3ec..353ac0fb7e2 100644
--- 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 class ManagedCursorImpl implements ManagedCursor {
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 class ManagedCursorImpl implements ManagedCursor {
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;