From f140a73549b889bcb81870dcfd6b6ca3a672be91 Mon Sep 17 00:00:00 2001 From: Nikhil Bharadwaj Ramashasthri Date: Sun, 16 Aug 2026 01:48:48 -0700 Subject: [PATCH] [fix][ml] Serialize durable cursor reset against concurrent acks under the pendingMarkDeleteOps monitor (#26304) A durable cursor reset stages its whole state mutation inside a single MarkDeleteEntry runnable, and internalFlushPendingMarkDeletes persists and runs only pendingMarkDeleteOps.getLast(). If an ack enqueues behind the reset entry it becomes getLast() and displaces the reset while triggerComplete still reports reset success. This serializes acks against the reset under the pendingMarkDeleteOps monitor. RESET_CURSOR_IN_PROGRESS is armed exactly where the reset enqueues its own entry in internalAsyncMarkDelete, so the flag being set is equivalent to the reset entry being committed to the queue. A concurrent ack that reaches the monitor before the reset entry is enqueued observes the flag unset, is accepted, and orders ahead of the reset, which stays getLast() and inherits the ack properties. An ack that arrives after observes the flag set and is rejected with a reset in progress error, so it cannot displace the reset. The rate limiter fast paths that skip the queue carry the same check through updateLastMarkDeleteEntryToLatest. Because the flag is armed at the enqueue rather than earlier, an ack already accepted before a reset orders ahead of it rather than being dropped, so a compaction cursor mark delete concurrent with a reset keeps its properties. Adds a deterministic test for the check before enqueue race and keeps the existing compaction property test green. --- .../mledger/impl/ManagedCursorImpl.java | 86 ++++++++-- .../mledger/impl/ManagedCursorTest.java | 156 +++++++++++++++++- 2 files changed, 224 insertions(+), 18 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 05446b7ef1c3e..95092a2591fa9 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 @@ -1604,8 +1604,14 @@ protected void internalResetCursor(Position proposedReadPosition, .attr("newReadPosition", newReadPosition) .log("Initiate reset readPosition"); + // Reject a second reset while one is already in progress. Reset operations for a cursor are serialized on the + // ledger executor (a single thread chosen by ledger name), and the RESET_CURSOR_IN_PROGRESS flag is only + // cleared under this monitor when the in-progress reset finishes, so a plain read here reliably detects it. + // The flag itself is armed later, atomically with enqueuing the reset entry in internalAsyncMarkDelete, so + // that an ack accepted before the reset entry is enqueued can still order ahead of the reset (and, for a + // compaction cursor, have its mark-delete properties carried forward by the reset) instead of being rejected. synchronized (pendingMarkDeleteOps) { - if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(this, FALSE, TRUE)) { + if (RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE) { log.error() .attr("readPosition", newReadPosition) .log("Reset requested, previous reset in progress"); @@ -1717,8 +1723,12 @@ public void operationFailed(ManagedLedgerException exception) { }; - persistentMarkDeletePosition = null; - inProgressMarkDeletePersistPosition = null; + // Arm RESET_CURSOR_IN_PROGRESS and enqueue the reset's own mark-delete entry atomically inside + // internalAsyncMarkDelete (isCursorReset=true). Arming the flag exactly at the enqueue - rather than earlier - + // means a concurrent ack that reaches internalAsyncMarkDelete before the reset entry is enqueued observes the + // flag still FALSE, is accepted, and orders ahead of the reset entry (so it cannot displace the reset, and the + // reset inherits its properties as the last pending entry); an ack that arrives after observes the flag TRUE + // and is rejected. Either way the reset entry stays pendingMarkDeleteOps.getLast(). internalAsyncMarkDelete(newMarkDeletePosition, isCompactionCursor() ? null : Collections.emptyMap(), new MarkDeleteCallback() { @Override @@ -1730,7 +1740,7 @@ public void markDeleteComplete(Object ctx) { public void markDeleteFailed(ManagedLedgerException exception, Object ctx) { finalCallback.operationFailed(exception); } - }, null, alignAcknowledgeStatusAfterPersisted); + }, null, alignAcknowledgeStatusAfterPersisted, true); } @Override @@ -2275,15 +2285,6 @@ public void asyncMarkDelete(final Position position, Map propertie return; } - if (RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE) { - log.debug().attr("position", position).log("Cursor reset in progress, ignoring mark delete"); - callback.markDeleteFailed( - new ManagedLedgerException("Reset cursor in progress - unable to mark delete position " - + position.toString()), - ctx); - return; - } - log.debug().attr("position", position).log("Mark delete"); Position newPosition = ackBatchPosition(position); @@ -2328,8 +2329,15 @@ public void asyncMarkDelete(final Position position, Map propertie // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { + // This fast path skips the pendingMarkDeleteOps enqueue, so it carries its own reset-in-progress guard: + // updateLastMarkDeleteEntryToLatest rejects (returns false) while a reset is in progress rather than + // racing the reset's rewind of lastMarkDeleteEntry. + if (!updateLastMarkDeleteEntryToLatest(newPosition, properties)) { + callback.markDeleteFailed(new ManagedLedgerException( + "Reset cursor in progress - unable to mark delete position " + newPosition), ctx); + return; + } isDirty = true; - updateLastMarkDeleteEntryToLatest(newPosition, properties); callback.markDeleteComplete(ctx); return; } @@ -2363,10 +2371,40 @@ private Position ackBatchPosition(Position position) { protected void internalAsyncMarkDelete(final Position newPosition, Map properties, final MarkDeleteCallback callback, final Object ctx, Runnable alignAcknowledgeStatusAfterPersisted) { + internalAsyncMarkDelete(newPosition, properties, callback, ctx, alignAcknowledgeStatusAfterPersisted, false); + } + + protected void internalAsyncMarkDelete(final Position newPosition, Map properties, + final MarkDeleteCallback callback, final Object ctx, Runnable alignAcknowledgeStatusAfterPersisted, + final boolean isCursorReset) { ledger.mbean.addMarkDeleteOp(); // We cannot write to the ledger during the switch, need to wait until the new metadata ledger is available synchronized (pendingMarkDeleteOps) { + // A reset stages its entire state mutation (markDeletePosition, readPosition, individualDeletedMessages, + // messagesConsumedCounter, ...) inside the alignAcknowledgeStatusAfterPersisted runnable of a single + // MarkDeleteEntry, and internalFlushPendingMarkDeletes only persists and runs the runnable of + // pendingMarkDeleteOps.getLast(). If a regular ack were enqueued behind the reset entry it would become + // getLast() and displace the reset: the ack's position would be persisted, the reset's runnable silently + // dropped, yet triggerComplete would still fire the reset callback, so resetComplete would report success + // for a reset that never took effect. + if (isCursorReset) { + // Arm the reset guard exactly here, atomically with appending the reset entry under this monitor. + // Any ack that already reached this monitor observed the flag FALSE, was accepted, and now sits ahead + // of the reset entry (the reset entry becomes getLast() and, via the last-pending-entry fallback + // below, carries that ack's properties forward). Any ack that arrives after this observes the flag + // TRUE and is rejected, so it cannot be enqueued behind the reset entry and displace it. + RESET_CURSOR_IN_PROGRESS_UPDATER.set(this, TRUE); + persistentMarkDeletePosition = null; + inProgressMarkDeletePersistPosition = null; + } else if (RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE) { + log.debug().attr("position", newPosition).log("Cursor reset in progress, rejecting mark delete"); + callback.markDeleteFailed( + new ManagedLedgerException("Reset cursor in progress - unable to mark delete position " + + newPosition), ctx); + return; + } + // use given properties or when missing, use the properties from the previous field value MarkDeleteEntry last = pendingMarkDeleteOps.peekLast(); Map propertiesToUse = @@ -2700,8 +2738,15 @@ public void asyncDelete(Iterable positions, AsyncCallbacks.DeleteCallb // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { + // This fast path skips the pendingMarkDeleteOps enqueue, so it carries its own reset-in-progress guard: + // updateLastMarkDeleteEntryToLatest rejects (returns false) while a reset is in progress rather than + // racing the reset's rewind of lastMarkDeleteEntry. + if (!updateLastMarkDeleteEntryToLatest(newMarkDeletePosition, null)) { + callback.deleteFailed(new ManagedLedgerException( + "Reset cursor in progress - unable to delete positions " + positions), ctx); + return; + } isDirty = true; - updateLastMarkDeleteEntryToLatest(newMarkDeletePosition, null); callback.deleteComplete(ctx); return; } @@ -2731,10 +2776,16 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) { } } - // update lastMarkDeleteEntry field if newPosition is later than the current lastMarkDeleteEntry.newPosition - private void updateLastMarkDeleteEntryToLatest(final Position newPosition, + // Update lastMarkDeleteEntry field if newPosition is later than the current lastMarkDeleteEntry.newPosition. + // Returns false without updating when a cursor reset is in progress, so the caller can reject the ack instead of + // racing the reset's rewind of lastMarkDeleteEntry. The reset-in-progress check is performed under the same + // pendingMarkDeleteOps monitor that guards the enqueue path, keeping it consistent with internalAsyncMarkDelete. + private boolean updateLastMarkDeleteEntryToLatest(final Position newPosition, final Map properties) { synchronized (pendingMarkDeleteOps) { + if (RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE) { + return false; + } // use given properties or when missing, use the properties from the previous field value MarkDeleteEntry lastPending = pendingMarkDeleteOps.peekLast(); Map propertiesToUse = @@ -2747,6 +2798,7 @@ private void updateLastMarkDeleteEntryToLatest(final Position newPosition, return new MarkDeleteEntry(newPosition, propertiesToUse, null, null); } }); + return true; } } 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 a6a0dc0411b03..1b210e5d25600 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 @@ -83,6 +83,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -6325,6 +6326,157 @@ public void close() { }).toList(), IntStream.range(0, 10).mapToObj(i -> "msg-" + i).toList()); } + /** + * Reproduces the silent-success reset drop from issue #26304 through the exact check-then-act race that + * maintainer review flagged, and asserts the concurrent delete is rejected with the reset-in-progress error. + * + *

A durable cursor reset stages its whole state mutation in a MarkDeleteEntry runnable, and + * internalFlushPendingMarkDeletes only persists and runs the runnable of pendingMarkDeleteOps.getLast(). If an + * individual delete is queued behind the reset entry while a read holds the mark-delete queue open, the delete + * becomes getLast() and displaces the reset: the delete's position is persisted, the reset's runnable is + * dropped, yet the reset callback still fires and reports success. + * + *

The old guard read RESET_CURSOR_IN_PROGRESS at the top of asyncDelete, before enqueuing - a check-then-act + * that a delete can pass while the flag is still FALSE, only for a reset to arm the flag and enqueue its entry + * before the delete reaches the queue. This test drives precisely that interleaving deterministically: it parks + * the delete on the cursor's write lock immediately after the point where the old check ran (the delete has + * therefore already "passed" the old check with the flag FALSE), then lets the reset arm the flag and enqueue, + * then releases the delete so it reaches the enqueue only afterwards. It fails on the pre-fix code (the delete + * is enqueued behind the reset and clobbers it, and the delete reports success) and passes once the check is + * decided atomically with the enqueue under the pendingMarkDeleteOps monitor. + */ + @Test(timeOut = 30000) + public void testResetCursorNotDroppedByConcurrentDelete() throws Exception { + // Hold the first data read open so PENDING_READ_OPS stays > 0 and mark-delete ops queue instead of + // being applied immediately. + AtomicBoolean armed = new AtomicBoolean(false); + AtomicReference> heldRead = new AtomicReference<>(); + AtomicReference heldReadEntries = new AtomicReference<>(); + bkc.setReadHandleInterceptor((ledgerId, firstEntry, lastEntry, entries) -> { + if (armed.compareAndSet(true, false)) { + CompletableFuture future = new CompletableFuture<>(); + heldReadEntries.set(entries); + heldRead.set(future); + return future; + } + return CompletableFuture.completedFuture(entries); + }); + + @Cleanup + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("testResetCursorNotDroppedByConcurrentDelete"); + ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1"); + cursor.setInactive(); // disable caching so reads hit the mock bookie interceptor + + List positions = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + positions.add(ledger.addEntry(("p-" + i).getBytes(Encoding))); + } + Position p0 = positions.get(0); + Position p5 = positions.get(5); + cursor.markDelete(positions.get(4)); // markDeletePosition = p4, readPosition = p5 + + // 1) Start a read and hold it so PENDING_READ_OPS > 0. + armed.set(true); + cursor.asyncReadEntries(1, new ReadEntriesCallback() { + @Override + public void readEntriesComplete(List entries, Object ctx) { + entries.forEach(Entry::release); + } + + @Override + public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { + } + }, null, PositionFactory.LATEST); + Awaitility.await().until(() -> cursor.getPendingReadOpsCount() == 1 && heldRead.get() != null); + + // 2) Park the concurrent delete right after the point where the old reset-in-progress check ran. Holding + // the cursor's write lock blocks asyncDelete at its lock.writeLock().lock() (the first blocking point + // after the removed early check), so on the pre-fix code the delete has already passed that check with + // the flag still FALSE - exactly the check-then-act window under test. + ReentrantReadWriteLock cursorLock = (ReentrantReadWriteLock) cursor.lock; + CountDownLatch deleteCallReturned = new CountDownLatch(1); + CountDownLatch deleteCallbackFired = new CountDownLatch(1); + AtomicBoolean deleteCompleted = new AtomicBoolean(false); + AtomicReference deleteFailure = new AtomicReference<>(); + DeleteCallback deleteCallback = new DeleteCallback() { + @Override + public void deleteComplete(Object ctx) { + deleteCompleted.set(true); + deleteCallbackFired.countDown(); + } + + @Override + public void deleteFailed(ManagedLedgerException exception, Object ctx) { + deleteFailure.set(exception); + deleteCallbackFired.countDown(); + } + }; + + CountDownLatch resetCompleted = new CountDownLatch(1); + AtomicBoolean resetSucceeded = new AtomicBoolean(false); + + cursorLock.writeLock().lock(); + Thread deleteThread; + try { + deleteThread = new Thread(() -> { + cursor.asyncDelete(p5, deleteCallback, null); + deleteCallReturned.countDown(); + }, "concurrent-delete"); + deleteThread.start(); + // Wait until the delete thread is blocked acquiring the write lock (it has passed the old check). + Awaitility.await().until(cursorLock::hasQueuedThreads); + + // 3) With the delete parked, reset the cursor back to p0. asyncResetCursor hops to the ledger executor + // and, because a read is pending, arms RESET_CURSOR_IN_PROGRESS and queues its reset entry. The + // reset never needs the write lock before enqueuing, so it completes this while the delete is parked. + cursor.asyncResetCursor(p0, false, new AsyncCallbacks.ResetCursorCallback() { + @Override + public void resetComplete(Object ctx) { + resetSucceeded.set(true); + resetCompleted.countDown(); + } + + @Override + public void resetFailed(ManagedLedgerException exception, Object ctx) { + resetCompleted.countDown(); + } + }); + Awaitility.await().untilAsserted(() -> { + synchronized (cursor.pendingMarkDeleteOps) { + assertEquals(cursor.pendingMarkDeleteOps.size(), 1); + assertEquals(cursor.pendingMarkDeleteOps.getLast().newPosition, ledger.getPreviousPosition(p0)); + } + }); + } finally { + // 4) Release the delete: it now reaches the enqueue only after the reset armed the flag and queued its + // entry. On the pre-fix code it is appended behind the reset (becoming getLast()); with the fix the + // atomic check under the pendingMarkDeleteOps monitor rejects it. + cursorLock.writeLock().unlock(); + } + assertTrue(deleteCallReturned.await(15, TimeUnit.SECONDS)); + + // 5) Release the held read. readOperationCompleted flushes the pending mark-delete queue and the reset + // persists and completes. + heldRead.get().complete(heldReadEntries.get()); + assertTrue(resetCompleted.await(15, TimeUnit.SECONDS)); + assertTrue(deleteCallbackFired.await(15, TimeUnit.SECONDS)); + deleteThread.join(TimeUnit.SECONDS.toMillis(15)); + + // The reset reported completion and must also have actually taken effect. + assertTrue(resetSucceeded.get(), "reset should have completed"); + assertEquals(cursor.getReadPosition(), p0, "reset readPosition was silently dropped"); + assertEquals(cursor.getMarkDeletedPosition(), ledger.getPreviousPosition(p0), + "reset markDeletePosition was silently dropped"); + assertFalse(cursor.isMessageDeleted(p5), "concurrent delete displaced the reset and persisted instead"); + + // The concurrent delete must be rejected with the reset-in-progress error, not silently reported complete + // (a shared latch that counts down on either outcome would hide this). + assertFalse(deleteCompleted.get(), "concurrent delete must not report success while a reset is in progress"); + assertNotNull(deleteFailure.get(), "concurrent delete should have failed with reset-in-progress"); + assertTrue(deleteFailure.get().getMessage().contains("Reset cursor in progress"), + "unexpected delete failure: " + deleteFailure.get().getMessage()); + } + @Test public void testSkipOpenLedgerFullyAcked() throws Exception { ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig(); @@ -6706,8 +6858,10 @@ public void testCompactionCursorResetNeverLoseMarkDeleteProperties() throws Exce } return invocation.callRealMethod(); + // Stub the 6-arg overload that both paths funnel through: the compaction mark-delete reaches it via the + // 5-arg overload's delegation (isCursorReset=false) and the reset calls it directly (isCursorReset=true). }).when(spyCursor).internalAsyncMarkDelete(any(Position.class), nullable(Map.class), - any(MarkDeleteCallback.class), nullable(Object.class), nullable(Runnable.class)); + any(MarkDeleteCallback.class), nullable(Object.class), nullable(Runnable.class), anyBoolean()); // Start compaction mark-delete from another thread because the spy intentionally blocks it. CompletableFuture.runAsync(() -> spyCursor.asyncMarkDelete(