Skip to content

[ML] Follow up on race condition fixes in ManagedCursorImpl #15031 #15067

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

Merged
lhotari merged 4 commits into from
Apr 14, 2022
Merged

[ML] Follow up on race condition fixes in ManagedCursorImpl #15031 #15067

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1e05c08
[ML] Fix race in persisting mark delete position
lhotari Apr 6, 2022
12f03dd
[ML] Resetting should reset lastMarkDeleteEntry
lhotari Apr 6, 2022
9344d99
[ML] Reset fields in initializeCursorPosition method
lhotari Apr 12, 2022
1b89b52
Address review feedback: latest -> later in log message
lhotari Apr 13, 2022
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
97 changes: 73 additions & 24 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 @@ -123,6 +123,11 @@ public class ManagedCursorImpl implements ManagedCursor {

// this position is have persistent mark delete position
protected volatile PositionImpl persistentMarkDeletePosition;
protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, PositionImpl>
INPROGRESS_MARKDELETE_PERSIST_POSITION_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, PositionImpl.class,
"inProgressMarkDeletePersistPosition");
protected volatile PositionImpl inProgressMarkDeletePersistPosition;

protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, PositionImpl> READ_POSITION_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, PositionImpl.class, "readPosition");
Expand Down Expand Up @@ -214,6 +219,31 @@ public MarkDeleteEntry(PositionImpl newPosition, Map<String, Long> properties,
this.callback = callback;
this.ctx = ctx;
}

public void triggerComplete() {
// Trigger the final callback after having (eventually) triggered the switchin-ledger operation. This
// will ensure that no race condition will happen between the next mark-delete and the switching
// operation.
if (callbackGroup != null) {
// Trigger the callback for every request in the group
for (MarkDeleteEntry e : callbackGroup) {
e.callback.markDeleteComplete(e.ctx);
}
} else if (callback != null) {
// Only trigger the callback for the current request
callback.markDeleteComplete(ctx);
}
}

public void triggerFailed(ManagedLedgerException exception) {
if (callbackGroup != null) {
for (MarkDeleteEntry e : callbackGroup) {
e.callback.markDeleteFailed(exception, e.ctx);
}
} else if (callback != null) {
callback.markDeleteFailed(exception, ctx);
}
}
}

protected final ArrayDeque<MarkDeleteEntry> pendingMarkDeleteOps = new ArrayDeque<>();
Expand Down Expand Up @@ -541,6 +571,7 @@ private void recoveredCursor(PositionImpl position, Map<String, Long> properties
messagesConsumedCounter = -getNumberOfEntries(Range.openClosed(position, ledger.getLastPosition()));
markDeletePosition = position;
persistentMarkDeletePosition = position;
inProgressMarkDeletePersistPosition = null;
readPosition = ledger.getNextValidPosition(position);
lastMarkDeleteEntry = new MarkDeleteEntry(markDeletePosition, properties, null, null);
// assign cursor-ledger so, it can be deleted when new ledger will be switched
Expand Down Expand Up @@ -1123,6 +1154,9 @@ public void operationFailed(ManagedLedgerException exception) {

};

persistentMarkDeletePosition = null;
inProgressMarkDeletePersistPosition = null;
lastMarkDeleteEntry = new MarkDeleteEntry(newPosition, getProperties(), null, null);
internalAsyncMarkDelete(newPosition, isCompactionCursor() ? getProperties() : Collections.emptyMap(),
new MarkDeleteCallback() {
@Override
Expand Down Expand Up @@ -1579,6 +1613,9 @@ boolean hasMoreEntries(PositionImpl position) {
void initializeCursorPosition(Pair<PositionImpl, Long> lastPositionCounter) {
readPosition = ledger.getNextValidPosition(lastPositionCounter.getLeft());
markDeletePosition = lastPositionCounter.getLeft();
lastMarkDeleteEntry = new MarkDeleteEntry(markDeletePosition, getProperties(), null, null);
persistentMarkDeletePosition = null;
inProgressMarkDeletePersistPosition = null;

// Initialize the counter such that the difference between the messages written on the ML and the
// messagesConsumed is 0, to ensure the initial backlog count is 0.
Expand Down Expand Up @@ -1793,6 +1830,34 @@ protected void internalAsyncMarkDelete(final PositionImpl newPosition, Map<Strin
}

void internalMarkDelete(final MarkDeleteEntry mdEntry) {
if (persistentMarkDeletePosition != null
&& mdEntry.newPosition.compareTo(persistentMarkDeletePosition) < 0) {
if (log.isInfoEnabled()) {
log.info("Skipping updating mark delete position to {}. The persisted mark delete position {} "
+ "is later.", mdEntry.newPosition, persistentMarkDeletePosition);
}
mdEntry.triggerComplete();
return;
}

PositionImpl inProgressLatest = INPROGRESS_MARKDELETE_PERSIST_POSITION_UPDATER.updateAndGet(this, current -> {
if (current != null && current.compareTo(mdEntry.newPosition) > 0) {
return current;
} else {
return mdEntry.newPosition;
}
});

// if there's a newer or equal mark delete update in progress, skip it.
if (inProgressLatest != mdEntry.newPosition) {
if (log.isInfoEnabled()) {
log.info("Skipping updating mark delete position to {}. The mark delete position update "
+ "in progress {} is later.", mdEntry.newPosition, inProgressLatest);
}
mdEntry.triggerComplete();
Comment thread
lhotari marked this conversation as resolved.
return;
}

// The counter is used to mark all the pending mark-delete request that were submitted to BK and that are not
// yet finished. While we have outstanding requests we cannot close the current ledger, so the switch to new
// ledger is postponed to when the counter goes to 0.
Expand All @@ -1815,6 +1880,9 @@ public void operationComplete() {
mdEntry.newPosition);
}

INPROGRESS_MARKDELETE_PERSIST_POSITION_UPDATER.compareAndSet(ManagedCursorImpl.this,
mdEntry.newPosition, null);

Comment thread
lhotari marked this conversation as resolved.
// Remove from the individual deleted messages all the entries before the new mark delete
// point.
lock.writeLock().lock();
Expand All @@ -1828,11 +1896,7 @@ public void operationComplete() {
subMap.values().forEach(BitSetRecyclable::recycle);
subMap.clear();
}
if (persistentMarkDeletePosition == null
|| mdEntry.newPosition.compareTo(persistentMarkDeletePosition) > 0) {
persistentMarkDeletePosition = mdEntry.newPosition;
}

persistentMarkDeletePosition = mdEntry.newPosition;
} finally {
lock.writeLock().unlock();
}
Expand All @@ -1841,22 +1905,13 @@ public void operationComplete() {

decrementPendingMarkDeleteCount();

// Trigger the final callback after having (eventually) triggered the switchin-ledger operation. This
// will ensure that no race condition will happen between the next mark-delete and the switching
// operation.
if (mdEntry.callbackGroup != null) {
// Trigger the callback for every request in the group
for (MarkDeleteEntry e : mdEntry.callbackGroup) {
e.callback.markDeleteComplete(e.ctx);
}
} else {
// Only trigger the callback for the current request
mdEntry.callback.markDeleteComplete(mdEntry.ctx);
}
mdEntry.triggerComplete();
}

@Override
public void operationFailed(ManagedLedgerException exception) {
INPROGRESS_MARKDELETE_PERSIST_POSITION_UPDATER.compareAndSet(ManagedCursorImpl.this,
mdEntry.newPosition, null);
isDirty = true;
log.warn("[{}] Failed to mark delete position for cursor={} position={}", ledger.getName(),
ManagedCursorImpl.this, mdEntry.newPosition);
Expand All @@ -1867,13 +1922,7 @@ public void operationFailed(ManagedLedgerException exception) {

decrementPendingMarkDeleteCount();

if (mdEntry.callbackGroup != null) {
for (MarkDeleteEntry e : mdEntry.callbackGroup) {
e.callback.markDeleteFailed(exception, e.ctx);
}
} else {
mdEntry.callback.markDeleteFailed(exception, mdEntry.ctx);
}
mdEntry.triggerFailed(exception);
}
});
}
Expand Down