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 2026/10/23 10:08:49 UTC
(pulsar) 18/23: [fix][ml] Don't trim ledgers while cursors are lazily recovering (#20814)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d3bd6b6fa44e91977d14eb541d28c70ed97f8a90
Author: lifepuzzlefun <wj...@163.com>
AuthorDate: Fri Oct 23 09:44:41 2026 +0800
[fix][ml] Don't trim ledgers while cursors are lazily recovering (#20814)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit 4aa22863c231ca3afb5cbb14541741ef44506561)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 99 +++++++++------
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 136 +++++++++++++++++++++
2 files changed, 201 insertions(+), 34 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index b26fed3e4d45..13be9dd146f2 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -228,6 +228,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// uninitialized cursor future from the 1st request
final Map<String, CompletableFuture<ManagedCursor>> uninitializedCursors;
+ // Cursors whose lazy recovery failed: they are recovered again when opened, and until then they block trimming,
+ // since their mark-delete position is unknown
+ final Set<String> failedLazyRecoveryCursors;
+
final EntryCache entryCache;
private ScheduledFuture<?> timeoutTask;
@@ -453,6 +457,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
this.waitingCursors = Queues.newConcurrentLinkedQueue();
this.waitingEntryCallBacks = Queues.newConcurrentLinkedQueue();
this.uninitializedCursors = new HashMap<>();
+ this.failedLazyRecoveryCursors = new HashSet<>();
this.clock = config.getClock();
// Get the next rollover time. Add a random value upto 5% to avoid rollover multiple ledgers at the same time
@@ -815,38 +820,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
} else {
// Lazily recover cursors by put them to uninitializedCursors map.
for (final String cursorName : consumers) {
- log.debug().attr("cursorName", cursorName).log("Recovering cursor lazily");
- final ManagedCursorImpl cursor;
- cursor = createCursor(ManagedLedgerImpl.this.bookKeeper, cursorName);
- final long timestampStartRecoverCursor = System.currentTimeMillis();
- CompletableFuture<ManagedCursor> cursorRecoveryFuture = new CompletableFuture<>();
- uninitializedCursors.put(cursorName, cursorRecoveryFuture);
-
- cursor.recover(new VoidCallback() {
- @Override
- public void operationComplete() {
- log.info().attr("cursorName", cursorName)
- .attr("position", cursor.getMarkDeletedPosition())
- .attr("remaining", cursorCount.get() - 1)
- .attr("cost ms", System.currentTimeMillis() - timestampStartRecoverCursor)
- .log("Lazy recovery for cursor completed");
- cursor.setActive();
- synchronized (ManagedLedgerImpl.this) {
- addCursor(cursor);
- uninitializedCursors.remove(cursor.getName()).complete(cursor);
- }
- }
-
- @Override
- public void operationFailed(ManagedLedgerException exception) {
- log.warn().attr("cursorName", cursorName).exception(exception)
- .attr("cost ms", System.currentTimeMillis() - timestampStartRecoverCursor)
- .log("Lazy recovery for cursor failed");
- synchronized (ManagedLedgerImpl.this) {
- uninitializedCursors.remove(cursor.getName()).completeExceptionally(exception);
- }
- }
- });
+ recoverCursorLazily(cursorName);
}
// Complete ledger recovery.
callback.initializeComplete();
@@ -861,6 +835,42 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
});
}
+ private synchronized CompletableFuture<ManagedCursor> recoverCursorLazily(String cursorName) {
+ log.debug().attr("cursorName", cursorName).log("Recovering cursor lazily");
+ final ManagedCursorImpl cursor = createCursor(bookKeeper, cursorName);
+ final long timestampStartRecoverCursor = System.currentTimeMillis();
+ CompletableFuture<ManagedCursor> cursorRecoveryFuture = new CompletableFuture<>();
+ uninitializedCursors.put(cursorName, cursorRecoveryFuture);
+
+ cursor.recover(new VoidCallback() {
+ @Override
+ public void operationComplete() {
+ log.info().attr("cursorName", cursorName)
+ .attr("position", cursor.getMarkDeletedPosition())
+ .attr("cost ms", System.currentTimeMillis() - timestampStartRecoverCursor)
+ .log("Lazy recovery for cursor completed");
+ cursor.setActive();
+ synchronized (ManagedLedgerImpl.this) {
+ failedLazyRecoveryCursors.remove(cursorName);
+ addCursor(cursor);
+ uninitializedCursors.remove(cursorName).complete(cursor);
+ }
+ }
+
+ @Override
+ public void operationFailed(ManagedLedgerException exception) {
+ log.warn().attr("cursorName", cursorName).exception(exception)
+ .attr("cost ms", System.currentTimeMillis() - timestampStartRecoverCursor)
+ .log("Lazy recovery for cursor failed");
+ synchronized (ManagedLedgerImpl.this) {
+ failedLazyRecoveryCursors.add(cursorName);
+ uninitializedCursors.remove(cursorName).completeExceptionally(exception);
+ }
+ }
+ });
+ return cursorRecoveryFuture;
+ }
+
private void addCursor(ManagedCursorImpl cursor) {
Position positionForOrdering = null;
if (cursor.isDurable()) {
@@ -1192,8 +1202,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
return;
}
- if (uninitializedCursors.containsKey(cursorName)) {
- uninitializedCursors.get(cursorName).thenAccept(cursor -> callback.openCursorComplete(cursor, ctx))
+ // A cursor whose lazy recovery failed still has its position persisted: recover it again rather than
+ // replacing it with a new one
+ CompletableFuture<ManagedCursor> uninitializedCursor = failedLazyRecoveryCursors.contains(cursorName)
+ ? recoverCursorLazily(cursorName) : uninitializedCursors.get(cursorName);
+ if (uninitializedCursor != null) {
+ uninitializedCursor.thenAccept(cursor -> callback.openCursorComplete(cursor, ctx))
.exceptionally(ex -> {
callback.openCursorFailed(ManagedLedgerException
.getManagedLedgerException(FutureUtil.unwrapCompletionException(ex)), ctx);
@@ -3721,6 +3735,23 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
promise.completeExceptionally(new ManagedLedgerFencedException("Can't trim fenced ledger"));
return;
}
+ // The slowest position below only covers the cursors in the container: a cursor that is still being
+ // recovered (or created) isn't there yet, and its mark-delete position may be in any ledger
+ if (!uninitializedCursors.isEmpty()) {
+ log.debug().attr("cursors", uninitializedCursors.keySet())
+ .log("Deferring trimming until the cursors are initialized");
+ trimmerMutex.unlock();
+ scheduleDeferredTrimming(isTruncate, promise);
+ return;
+ }
+ if (!failedLazyRecoveryCursors.isEmpty()) {
+ log.debug().attr("cursors", failedLazyRecoveryCursors)
+ .log("Ignoring trimming request since cursors failed to recover");
+ trimmerMutex.unlock();
+ promise.completeExceptionally(new ManagedLedgerException(
+ "Can't trim ledger until cursors are recovered: " + failedLazyRecoveryCursors));
+ return;
+ }
// A terminated managed ledger has no current ledger once its final ledger was deleted from BookKeeper,
// and that final ledger is then kept like a current one
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index da6129044abc..2fc27515256c 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -77,10 +77,12 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -3592,6 +3594,140 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
assertEquals(cursor.getMarkDeletedPosition(), p1);
}
+ /**
+ * Returns a factory whose managed ledgers recover each cursor once the future of its name in
+ * {@code recoveryGates} completes, and fail its recovery if that future fails.
+ */
+ private ManagedLedgerFactoryImpl createFactoryWithGatedCursorRecovery(
+ Map<String, CompletableFuture<Void>> recoveryGates) throws Exception {
+ return new ManagedLedgerFactoryImpl(metadataStore, bkc) {
+ @Override
+ protected ManagedLedgerImpl createManagedLedger(BookKeeper bk, MetaStore store, String name,
+ ManagedLedgerConfig config,
+ Supplier<CompletableFuture<Boolean>> mlOwnershipChecker) {
+ return new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker) {
+ @Override
+ protected ManagedCursorImpl createCursor(BookKeeper bookKeeper, String cursorName) {
+ return new ManagedCursorImpl(bookKeeper, this, cursorName) {
+ @Override
+ void recover(VoidCallback callback) {
+ recoveryGates.get(cursorName).whenComplete((__, ex) -> {
+ if (ex != null) {
+ callback.operationFailed(new ManagedLedgerException(ex));
+ } else {
+ super.recover(callback);
+ }
+ });
+ }
+ };
+ }
+ };
+ }
+ };
+ }
+
+ @Test
+ public void testLazyCursorRecoveryDefersTrimming() throws Exception {
+ ManagedLedgerConfig config = initManagedLedgerConfig(defaultConfig()).setMaxEntriesPerLedger(2);
+ ManagedLedger ledger = factory.open("testLazyCursorRecoveryDefersTrimming", config);
+ ManagedCursor slowCursor = ledger.openCursor("slow-cursor");
+ ManagedCursor fastCursor = ledger.openCursor("fast-cursor");
+ Position lastPosition = null;
+ for (int i = 0; i < 6; i++) {
+ lastPosition = ledger.addEntry(("entry-" + i).getBytes());
+ }
+ Position slowPosition = slowCursor.getMarkDeletedPosition();
+ fastCursor.markDelete(lastPosition);
+ ledger.close();
+
+ Map<String, CompletableFuture<Void>> recoveryGates = new ConcurrentHashMap<>();
+ recoveryGates.put("slow-cursor", new CompletableFuture<>());
+ recoveryGates.put("fast-cursor", new CompletableFuture<>());
+ @Cleanup("shutdown")
+ ManagedLedgerFactoryImpl factory2 = createFactoryWithGatedCursorRecovery(recoveryGates);
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory2.open("testLazyCursorRecoveryDefersTrimming",
+ initManagedLedgerConfig(defaultConfig()).setMaxEntriesPerLedger(2).setLazyCursorRecovery(true));
+ List<Long> ledgerIds = new ArrayList<>(reopened.getLedgersInfo().keySet());
+
+ // No cursor recovered yet: trimming must not see the managed ledger as having no durable cursors
+ CompletableFuture<Void> trimmed = new CompletableFuture<>();
+ reopened.trimConsumedLedgersInBackground(trimmed);
+ Awaitility.await().during(500, TimeUnit.MILLISECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> !trimmed.isDone());
+ assertEquals(new ArrayList<>(reopened.getLedgersInfo().keySet()), ledgerIds);
+
+ // Only the fast cursor recovered: its position alone must not decide the trimming
+ recoveryGates.get("fast-cursor").complete(null);
+ assertEquals(reopened.openCursor("fast-cursor").getMarkDeletedPosition(), lastPosition);
+ Awaitility.await().during(500, TimeUnit.MILLISECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> !trimmed.isDone());
+ assertEquals(new ArrayList<>(reopened.getLedgersInfo().keySet()), ledgerIds);
+
+ // Once the slow cursor recovered, the deferred trimming completes and keeps the ledgers it needs
+ recoveryGates.get("slow-cursor").complete(null);
+ trimmed.get(5, TimeUnit.SECONDS);
+ ManagedCursor recoveredSlowCursor = reopened.openCursor("slow-cursor");
+ assertEquals(recoveredSlowCursor.getMarkDeletedPosition(), slowPosition);
+ assertEquals(new ArrayList<>(reopened.getLedgersInfo().keySet()), ledgerIds);
+ assertEquals(recoveredSlowCursor.getNumberOfEntriesInBacklog(false), 6);
+
+ // Trimming proceeds normally after the slow cursor acknowledges everything
+ recoveredSlowCursor.markDelete(lastPosition);
+ CompletableFuture<Void> trimmedAfterAck = new CompletableFuture<>();
+ reopened.trimConsumedLedgersInBackground(trimmedAfterAck);
+ trimmedAfterAck.get(5, TimeUnit.SECONDS);
+ assertTrue(reopened.getLedgersInfo().size() < ledgerIds.size());
+ }
+
+ @Test
+ public void testFailedLazyCursorRecoveryBlocksTrimmingUntilRecovered() throws Exception {
+ ManagedLedgerConfig config = initManagedLedgerConfig(defaultConfig()).setMaxEntriesPerLedger(2);
+ ManagedLedger ledger = factory.open("testFailedLazyCursorRecovery", config);
+ ManagedCursor cursor = ledger.openCursor("cursor");
+ Position lastPosition = null;
+ for (int i = 0; i < 6; i++) {
+ lastPosition = ledger.addEntry(("entry-" + i).getBytes());
+ }
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+ ledger.close();
+
+ Map<String, CompletableFuture<Void>> recoveryGates = new ConcurrentHashMap<>();
+ recoveryGates.put("cursor", CompletableFuture.failedFuture(new Exception("recovery failed")));
+ @Cleanup("shutdown")
+ ManagedLedgerFactoryImpl factory2 = createFactoryWithGatedCursorRecovery(recoveryGates);
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory2.open("testFailedLazyCursorRecovery",
+ initManagedLedgerConfig(defaultConfig()).setMaxEntriesPerLedger(2).setLazyCursorRecovery(true));
+ List<Long> ledgerIds = new ArrayList<>(reopened.getLedgersInfo().keySet());
+
+ // The failed cursor isn't known to the managed ledger, but its position still protects the ledgers
+ Awaitility.await().until(() -> {
+ synchronized (reopened) {
+ return reopened.uninitializedCursors.isEmpty();
+ }
+ });
+ CompletableFuture<Void> trimmed = new CompletableFuture<>();
+ reopened.trimConsumedLedgersInBackground(trimmed);
+ try {
+ trimmed.get(5, TimeUnit.SECONDS);
+ fail("Trimming should fail while a cursor isn't recovered");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof ManagedLedgerException);
+ }
+ assertEquals(new ArrayList<>(reopened.getLedgersInfo().keySet()), ledgerIds);
+
+ // Opening the cursor recovers it again instead of creating it at the end of the managed ledger
+ recoveryGates.put("cursor", CompletableFuture.completedFuture(null));
+ ManagedCursor recovered = reopened.openCursor("cursor");
+ assertEquals(recovered.getMarkDeletedPosition(), markDeletePosition);
+ assertEquals(recovered.getNumberOfEntriesInBacklog(false), 6);
+
+ recovered.markDelete(lastPosition);
+ CompletableFuture<Void> trimmedAfterAck = new CompletableFuture<>();
+ reopened.trimConsumedLedgersInBackground(trimmedAfterAck);
+ trimmedAfterAck.get(5, TimeUnit.SECONDS);
+ assertTrue(reopened.getLedgersInfo().size() < ledgerIds.size());
+ }
+
@Test
public void testConcurrentOpenCursor() throws Exception {
ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("testConcurrentOpenCursor",