Skip to content

[fix][ml] Prevent durable cursor reset from being silently discarded by a concurrent individual delete #26318

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nikhiln64 wants to merge 1 commit into
base: master
Choose a base branch
from
+224 −18
Open

[fix][ml] Prevent durable cursor reset from being silently discarded by a concurrent individual delete #26318

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 69 additions & 17 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -2275,15 +2285,6 @@ public void asyncMarkDelete(final Position position, Map<String, Long> 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);
Expand Down Expand Up @@ -2328,8 +2329,15 @@ public void asyncMarkDelete(final Position position, Map<String, Long> 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;
}
Expand Down Expand Up @@ -2363,10 +2371,40 @@ private Position ackBatchPosition(Position position) {

protected void internalAsyncMarkDelete(final Position newPosition, Map<String, Long> properties,
final MarkDeleteCallback callback, final Object ctx, Runnable alignAcknowledgeStatusAfterPersisted) {
internalAsyncMarkDelete(newPosition, properties, callback, ctx, alignAcknowledgeStatusAfterPersisted, false);
}

protected void internalAsyncMarkDelete(final Position newPosition, Map<String, Long> 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<String, Long> propertiesToUse =
Expand Down Expand Up @@ -2700,8 +2738,15 @@ public void asyncDelete(Iterable<Position> 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;
}
Expand Down Expand Up @@ -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<String, Long> 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<String, Long> propertiesToUse =
Expand All @@ -2747,6 +2798,7 @@ private void updateLastMarkDeleteEntryToLatest(final Position newPosition,
return new MarkDeleteEntry(newPosition, propertiesToUse, null, null);
}
});
return true;
}
}

Expand Down
Loading