You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ji...@apache.org on 2024/12/16 11:22:34 UTC
(pulsar) branch master updated: [fix][broker] Avoid block markDeletePosition forward when skip lost entries (#21210)
This is an automated email from the ASF dual-hosted git repository.
jianghaiting pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 3761dc482cf [fix][broker] Avoid block markDeletePosition forward when skip lost entries (#21210)
3761dc482cf is described below
commit 3761dc482cf0566ad012b06a908422938bc3f391
Author: hanmz <gu...@tencent.com>
AuthorDate: Mon Dec 16 19:22:28 2024 +0800
[fix][broker] Avoid block markDeletePosition forward when skip lost entries (#21210)
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
Co-authored-by: Lari Hotari <lh...@apache.org>
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 48 ++++++++++++++++++++++
.../bookkeeper/mledger/impl/OpReadEntry.java | 2 +
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 47 +++++++++++++++++++++
3 files changed, 97 insertions(+)
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 478c6a1b379..022cecf8d57 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
@@ -2890,6 +2890,54 @@ public class ManagedCursorImpl implements ManagedCursor {
}, null);
}
+ /**
+ * Manually acknowledge all entries from startPosition to endPosition.
+ * - Since this is an uncommon event, we focus on maintainability. So we do not modify
+ * {@link #individualDeletedMessages} and {@link #batchDeletedIndexes}, but call
+ * {@link #asyncDelete(Position, AsyncCallbacks.DeleteCallback, Object)}.
+ * - This method is valid regardless of the consumer ACK type.
+ * - If there is a consumer ack request after this event, it will also work.
+ */
+ public void skipNonRecoverableEntries(Position startPosition, Position endPosition){
+ long ledgerId = startPosition.getLedgerId();
+ LedgerInfo ledgerInfo = ledger.getLedgersInfo().get(ledgerId);
+ if (ledgerInfo == null) {
+ return;
+ }
+
+ long startEntryId = Math.max(0, startPosition.getEntryId());
+ long endEntryId = ledgerId != endPosition.getLedgerId() ? ledgerInfo.getEntries() : endPosition.getEntryId();
+ if (startEntryId >= endEntryId) {
+ return;
+ }
+
+ lock.writeLock().lock();
+ log.warn("[{}] [{}] Since these entry for ledger [{}] is lost and the autoSkipNonRecoverableData is true, "
+ + "these entries [{}:{}) will be auto acknowledge in subscription",
+ ledger.getName(), name, ledgerId, startEntryId, endEntryId);
+ try {
+ for (long i = startEntryId; i < endEntryId; i++) {
+ if (!individualDeletedMessages.contains(ledgerId, i)) {
+ asyncDelete(PositionFactory.create(ledgerId, i), new AsyncCallbacks.DeleteCallback() {
+ @Override
+ public void deleteComplete(Object ctx) {
+ // ignore.
+ }
+
+ @Override
+ public void deleteFailed(ManagedLedgerException ex, Object ctx) {
+ // The method internalMarkDelete already handled the failure operation. We only need to
+ // make sure the memory state is updated.
+ // If the broker crashed, the non-recoverable ledger will be detected again.
+ }
+ }, null);
+ }
+ }
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
// //////////////////////////////////////////////////
void startCreatingNewMetadataLedger() {
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
index 3fd7e36c433..a4928b44bd9 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
@@ -137,6 +137,8 @@ class OpReadEntry implements ReadEntriesCallback {
updateReadPosition(nexReadPosition);
if (lostLedger != null) {
cursor.getManagedLedger().skipNonRecoverableLedger(lostLedger);
+ } else {
+ cursor.skipNonRecoverableEntries(readPosition, nexReadPosition);
}
checkReadCompletion();
} else {
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 b4ab673facb..69b74fcf8f5 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
@@ -4766,6 +4766,53 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
&& cursorReadPosition.getEntryId() == expectReadPosition.getEntryId());
}
+ @Test
+ public void testSkipNonRecoverableEntries() throws ManagedLedgerException, InterruptedException {
+ ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+ int maxMessagePerLedger = 10;
+ managedLedgerConfig.setMaxEntriesPerLedger(maxMessagePerLedger);
+ ManagedLedger ledger = factory.open("testSkipNonRecoverableEntries", managedLedgerConfig);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("my-cursor");
+
+ Position lacPosition = ledger.getLastConfirmedEntry();
+ long ledgerId = lacPosition.getLedgerId();
+ assertEquals(PositionFactory.create(ledgerId, -1), cursor.getMarkDeletedPosition());
+
+ // Mock add 10 entry
+ for (int i = 0; i < 10; i++) {
+ ledger.addEntry(String.valueOf(i).getBytes());
+ }
+
+ // read 2 entry and delete these entries, MarkDeletedPosition move forward
+ List<Entry> entries = cursor.readEntries(2);
+ for (Entry entry : entries) {
+ cursor.delete(entry.getPosition());
+ }
+ assertEquals(PositionFactory.create(ledgerId, 1), cursor.getMarkDeletedPosition());
+
+ // read the next 6 entry and not delete, MarkDeletedPosition not move forward
+ entries = cursor.readEntries(6);
+ assertEquals(PositionFactory.create(ledgerId, 1), cursor.getMarkDeletedPosition());
+
+ // delete last read entry, MarkDeletedPosition not move forward
+ Entry lastEntry = entries.get(entries.size() - 1);
+ cursor.delete(lastEntry.getPosition());
+ assertEquals(PositionFactory.create(ledgerId, 1), cursor.getMarkDeletedPosition());
+
+ // call skip entries, MarkDeletedPosition move forward
+ cursor.skipNonRecoverableEntries(cursor.getMarkDeletedPosition(),
+ PositionFactory.create(ledgerId, lastEntry.getEntryId()));
+ assertEquals(PositionFactory.create(ledgerId, lastEntry.getEntryId()), cursor.getMarkDeletedPosition());
+
+ // repeat call skip entries, MarkDeletedPosition not change
+ cursor.skipNonRecoverableEntries(cursor.getMarkDeletedPosition(),
+ PositionFactory.create(ledgerId, lastEntry.getEntryId()));
+ assertEquals(PositionFactory.create(ledgerId, lastEntry.getEntryId()), cursor.getMarkDeletedPosition());
+
+ cursor.close();
+ ledger.close();
+ }
+
@Test
public void testRecoverCursorWithTerminateManagedLedger() throws Exception {
String mlName = "my_test_ledger";