You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2026/07/21 11:01:37 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Fix `getEstimatedSizeSinceMarkDeletePosition` throw `IllegalArgumentException` (#26184)
This is an automated email from the ASF dual-hosted git repository.
Technoboy- pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new d14d119e88a9 [fix][broker] Fix `getEstimatedSizeSinceMarkDeletePosition` throw `IllegalArgumentException` (#26184)
d14d119e88a9 is described below
commit d14d119e88a993cd33299e88929f9d0fc6ef4242
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Tue Jul 21 10:53:43 2026 +0800
[fix][broker] Fix `getEstimatedSizeSinceMarkDeletePosition` throw `IllegalArgumentException` (#26184)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 25 ++++++-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 84 ++++++++++++++++++++++
2 files changed, 107 insertions(+), 2 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 6da7b5308b36..7a2ceecbd3d0 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
@@ -1291,6 +1291,21 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public long getEstimatedSizeSinceMarkDeletePosition() {
+ Position markDeletePosition = this.markDeletePosition;
+ Position lastPosition = ledger.getLastPosition();
+ if (markDeletePosition == null || markDeletePosition.compareTo(lastPosition) == 0) {
+ return 0;
+ }
+ if (markDeletePosition.compareTo(lastPosition) > 0) {
+ if (!ledger.ledgerExists(lastPosition.getLedgerId())
+ || isMarkDeletePositionOnEmptyCurrentLedger(markDeletePosition)) {
+ return 0;
+ }
+ throw new IllegalArgumentException(String.format(
+ "Cursor %s mark-delete position %s is ahead of the last position %s for managed ledger %s",
+ name, markDeletePosition, lastPosition, ledger.getName()));
+ }
+
long totalSize = ledger.estimateBacklogFromPosition(markDeletePosition);
// Need to subtract size of individual deleted messages
@@ -1303,7 +1318,7 @@ public class ManagedCursorImpl implements ManagedCursor {
long deletedCount = 0;
lock.readLock().lock();
try {
- Range<Position> backlogRange = Range.openClosed(markDeletePosition, ledger.getLastPosition());
+ Range<Position> backlogRange = Range.openClosed(markDeletePosition, lastPosition);
if (getConfig().isUnackedRangesOpenCacheSetEnabled()) {
deletedCount = individualDeletedMessages.cardinality(
@@ -1330,7 +1345,7 @@ public class ManagedCursorImpl implements ManagedCursor {
}
// Estimate size by using average entry size from the backlog range
- Range<Position> backlogRange = Range.openClosed(markDeletePosition, ledger.getLastPosition());
+ Range<Position> backlogRange = Range.openClosed(markDeletePosition, lastPosition);
long totalEntriesInBacklog = ledger.getNumberOfEntries(backlogRange);
if (totalEntriesInBacklog <= deletedCount || totalEntriesInBacklog == 0) {
@@ -1356,6 +1371,12 @@ public class ManagedCursorImpl implements ManagedCursor {
return adjustedSize;
}
+ private boolean isMarkDeletePositionOnEmptyCurrentLedger(Position markDeletePosition) {
+ return ledger.currentLedger != null
+ && markDeletePosition.getLedgerId() == ledger.currentLedger.getId()
+ && ledger.currentLedgerEntries == 0;
+ }
+
private long getNumberOfEntriesInBacklog() {
if (markDeletePosition.compareTo(ledger.getLastPosition()) >= 0) {
return 0;
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index 70cd5b06ee2d..537bd845a778 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -90,6 +90,7 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.PulsarMockBookKeeper;
import org.apache.bookkeeper.client.PulsarMockReadHandleInterceptor;
import org.apache.bookkeeper.client.api.LedgerEntries;
@@ -3797,6 +3798,89 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 10 * entryData.length);
}
+ @Test
+ public void testEstimatedUnackedSizeWhenLatestLedgerIsEmpty() throws Exception {
+ ManagedLedger ledger = factory.open("test_estimated_unacked_size_empty_latest_ledger");
+ ManagedCursor cursor = ledger.openCursor("c1");
+
+ assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 0);
+ }
+
+ @Test
+ public void testEstimatedUnackedSizeWhenCursorCaughtUpWithLastPosition() throws Exception {
+ ManagedLedger ledger = factory.open("test_estimated_unacked_size_cursor_caught_up");
+ ManagedCursor cursor = ledger.openCursor("c1");
+
+ Position lastPosition = ledger.addEntry("entry".getBytes(Encoding));
+ cursor.markDelete(lastPosition);
+
+ assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 0);
+ }
+
+ @Test
+ public void testEstimatedUnackedSizeWhenCursorAdvancedToEmptyCurrentLedger() {
+ ManagedLedgerImpl ledger = mock(ManagedLedgerImpl.class);
+ when(ledger.getName()).thenReturn("test_estimated_unacked_size_empty_current_ledger");
+ when(ledger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(ledger.getLogger()).thenReturn(log);
+
+ long currentLedgerId = 4;
+ Position lastPosition = PositionFactory.create(3, 9);
+ Position markDeletePosition = PositionFactory.create(currentLedgerId, -1);
+ when(ledger.getLastPosition()).thenReturn(lastPosition);
+ when(ledger.ledgerExists(lastPosition.getLedgerId())).thenReturn(true);
+
+ LedgerHandle currentLedger = mock(LedgerHandle.class);
+ when(currentLedger.getId()).thenReturn(currentLedgerId);
+ ledger.currentLedger = currentLedger;
+ ledger.currentLedgerEntries = 0;
+
+ ManagedCursorImpl cursor = new ManagedCursorImpl(mock(BookKeeper.class), ledger, "c1");
+ cursor.markDeletePosition = markDeletePosition;
+
+ assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 0);
+ verify(ledger, never()).estimateBacklogFromPosition(any());
+ }
+
+ @Test
+ public void testEstimatedUnackedSizeWhenLastPositionLedgerIsNoLongerInLedgerList() {
+ ManagedLedgerImpl ledger = mock(ManagedLedgerImpl.class);
+ when(ledger.getName()).thenReturn("test_estimated_unacked_size_last_position_ledger_removed");
+ when(ledger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(ledger.getLogger()).thenReturn(log);
+
+ Position lastPosition = PositionFactory.create(3, 0);
+ Position markDeletePosition = PositionFactory.create(4, -1);
+ when(ledger.getLastPosition()).thenReturn(lastPosition);
+ when(ledger.ledgerExists(lastPosition.getLedgerId())).thenReturn(false);
+
+ ManagedCursorImpl cursor = new ManagedCursorImpl(mock(BookKeeper.class), ledger, "c1");
+ cursor.markDeletePosition = markDeletePosition;
+
+ assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 0);
+ }
+
+ @Test
+ public void testEstimatedUnackedSizeFailsWhenCursorIsUnexpectedlyAheadOfLastPosition() {
+ ManagedLedgerImpl ledger = mock(ManagedLedgerImpl.class);
+ when(ledger.getName()).thenReturn("test_estimated_unacked_size_unexpected_position");
+ when(ledger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(ledger.getLogger()).thenReturn(log);
+
+ Position lastPosition = PositionFactory.create(1, 10);
+ Position markDeletePosition = PositionFactory.create(2, -1);
+ when(ledger.getLastPosition()).thenReturn(lastPosition);
+ when(ledger.ledgerExists(lastPosition.getLedgerId())).thenReturn(true);
+ when(ledger.ledgerExists(markDeletePosition.getLedgerId())).thenReturn(true);
+
+ ManagedCursorImpl cursor = new ManagedCursorImpl(mock(BookKeeper.class), ledger, "c1");
+ cursor.markDeletePosition = markDeletePosition;
+
+ IllegalArgumentException exception = Assert.expectThrows(IllegalArgumentException.class,
+ cursor::getEstimatedSizeSinceMarkDeletePosition);
+ assertTrue(exception.getMessage().contains("is ahead of the last position"));
+ }
+
/**
* Test that cursor.getEstimatedSizeSinceMarkDeletePosition() correctly accounts for individual
* message deletions (asyncDelete/individual ack).