From 6479d835efa29ec4b2610a01ce820b3758d1d0b8 Mon Sep 17 00:00:00 2001 From: hanmz Date: Wed, 20 Sep 2023 21:43:36 +0800 Subject: [PATCH 1/3] Delete skip entries for moving markDeletePosition forward. --- .../mledger/impl/ManagedCursorImpl.java | 48 +++++++++++++++++++ .../bookkeeper/mledger/impl/OpReadEntry.java | 2 + 2 files changed, 50 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 376bde3dc2cac..d4706a120acd4 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 @@ -2791,6 +2791,54 @@ public void deleteFailed(ManagedLedgerException ex, Object ctx) { } } + /** + * 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(PositionImpl.get(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 7b59c3903d5bc..eac2b85da5eaf 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 @@ -135,6 +135,8 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { updateReadPosition(nexReadPosition); if (lostLedger != null) { cursor.getManagedLedger().skipNonRecoverableLedger(lostLedger); + } else { + cursor.skipNonRecoverableEntries(readPosition, nexReadPosition); } checkReadCompletion(); } else { From a00cbb869088cef52c2309784c07876a6a67d45f Mon Sep 17 00:00:00 2001 From: hanmz Date: Thu, 28 Sep 2023 18:18:13 +0800 Subject: [PATCH 2/3] [fix][broker] release orphan replicator after replicator is closed. --- .../mledger/impl/ManagedCursorTest.java | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) 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 627ae73d928bd..f6e89aa561117 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 @@ -4484,5 +4484,52 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { ledger.close(); } + @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(PositionImpl.get(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 entries = cursor.readEntries(2); + for (Entry entry : entries) { + cursor.delete(entry.getPosition()); + } + assertEquals(PositionImpl.get(ledgerId, 1), cursor.getMarkDeletedPosition()); + + // read the next 6 entry and not delete, MarkDeletedPosition not move forward + entries = cursor.readEntries(6); + assertEquals(PositionImpl.get(ledgerId, 1), cursor.getMarkDeletedPosition()); + + // delete last read entry, MarkDeletedPosition not move forward + Entry lastEntry = entries.get(entries.size() - 1); + cursor.delete(lastEntry.getPosition()); + assertEquals(PositionImpl.get(ledgerId, 1), cursor.getMarkDeletedPosition()); + + // call skip entries, MarkDeletedPosition move forward + cursor.skipNonRecoverableEntries(cursor.getMarkDeletedPosition(), + PositionImpl.get(ledgerId, lastEntry.getEntryId())); + assertEquals(PositionImpl.get(ledgerId, lastEntry.getEntryId()), cursor.getMarkDeletedPosition()); + + // repeat call skip entries, MarkDeletedPosition not change + cursor.skipNonRecoverableEntries(cursor.getMarkDeletedPosition(), + PositionImpl.get(ledgerId, lastEntry.getEntryId())); + assertEquals(PositionImpl.get(ledgerId, lastEntry.getEntryId()), cursor.getMarkDeletedPosition()); + + cursor.close(); + ledger.close(); + } + private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class); } From 7b95ac81b9c857a6dbb1140ce4de1046026f3ec0 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 9 Oct 2024 17:42:48 +0300 Subject: [PATCH 3/3] Adapt to master branch --- .../mledger/impl/ManagedCursorImpl.java | 2 +- .../mledger/impl/ManagedCursorTest.java | 20 +++++++++---------- 2 files changed, 11 insertions(+), 11 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 67283ec51ebad..fa019d7b58a40 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 @@ -2918,7 +2918,7 @@ public void skipNonRecoverableEntries(Position startPosition, Position endPositi try { for (long i = startEntryId; i < endEntryId; i++) { if (!individualDeletedMessages.contains(ledgerId, i)) { - asyncDelete(PositionImpl.get(ledgerId, i), new AsyncCallbacks.DeleteCallback() { + asyncDelete(PositionFactory.create(ledgerId, i), new AsyncCallbacks.DeleteCallback() { @Override public void deleteComplete(Object ctx) { // ignore. 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 ca122bf3b5012..b52e3061eef2e 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 @@ -75,8 +75,8 @@ import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.BookKeeper.DigestType; import org.apache.bookkeeper.client.LedgerEntry; -import org.apache.bookkeeper.client.api.ReadHandle; import org.apache.bookkeeper.client.PulsarMockBookKeeper; +import org.apache.bookkeeper.client.api.ReadHandle; import org.apache.bookkeeper.common.util.OrderedExecutor; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback; @@ -4776,7 +4776,7 @@ public void testSkipNonRecoverableEntries() throws ManagedLedgerException, Inter Position lacPosition = ledger.getLastConfirmedEntry(); long ledgerId = lacPosition.getLedgerId(); - assertEquals(PositionImpl.get(ledgerId, -1), cursor.getMarkDeletedPosition()); + assertEquals(PositionFactory.create(ledgerId, -1), cursor.getMarkDeletedPosition()); // Mock add 10 entry for (int i = 0; i < 10; i++) { @@ -4788,31 +4788,31 @@ public void testSkipNonRecoverableEntries() throws ManagedLedgerException, Inter for (Entry entry : entries) { cursor.delete(entry.getPosition()); } - assertEquals(PositionImpl.get(ledgerId, 1), cursor.getMarkDeletedPosition()); + assertEquals(PositionFactory.create(ledgerId, 1), cursor.getMarkDeletedPosition()); // read the next 6 entry and not delete, MarkDeletedPosition not move forward entries = cursor.readEntries(6); - assertEquals(PositionImpl.get(ledgerId, 1), cursor.getMarkDeletedPosition()); + 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(PositionImpl.get(ledgerId, 1), cursor.getMarkDeletedPosition()); + assertEquals(PositionFactory.create(ledgerId, 1), cursor.getMarkDeletedPosition()); // call skip entries, MarkDeletedPosition move forward cursor.skipNonRecoverableEntries(cursor.getMarkDeletedPosition(), - PositionImpl.get(ledgerId, lastEntry.getEntryId())); - assertEquals(PositionImpl.get(ledgerId, lastEntry.getEntryId()), 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(), - PositionImpl.get(ledgerId, lastEntry.getEntryId())); - assertEquals(PositionImpl.get(ledgerId, lastEntry.getEntryId()), 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";