From 1e05c08859c2e8033bb678cc13dc55750aa465f7 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 6 Apr 2022 15:57:20 +0300 Subject: [PATCH 1/4] [ML] Fix race in persisting mark delete position --- .../mledger/impl/ManagedCursorImpl.java | 93 ++++++++++++++----- 1 file changed, 69 insertions(+), 24 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 199868b543fdd..77b0aaeb9be1d 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 @@ -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 + INPROGRESS_MARKDELETE_PERSIST_POSITION_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, PositionImpl.class, + "inProgressMarkDeletePersistPosition"); + protected volatile PositionImpl inProgressMarkDeletePersistPosition; protected static final AtomicReferenceFieldUpdater READ_POSITION_UPDATER = AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, PositionImpl.class, "readPosition"); @@ -214,6 +219,31 @@ public MarkDeleteEntry(PositionImpl newPosition, Map 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 pendingMarkDeleteOps = new ArrayDeque<>(); @@ -541,6 +571,7 @@ private void recoveredCursor(PositionImpl position, Map 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 @@ -1123,6 +1154,8 @@ public void operationFailed(ManagedLedgerException exception) { }; + persistentMarkDeletePosition = null; + inProgressMarkDeletePersistPosition = null; internalAsyncMarkDelete(newPosition, isCompactionCursor() ? getProperties() : Collections.emptyMap(), new MarkDeleteCallback() { @Override @@ -1793,6 +1826,34 @@ protected void internalAsyncMarkDelete(final PositionImpl newPosition, Map { + 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 latest.", mdEntry.newPosition, inProgressLatest); + } + mdEntry.triggerComplete(); + 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. @@ -1815,6 +1876,9 @@ public void operationComplete() { mdEntry.newPosition); } + INPROGRESS_MARKDELETE_PERSIST_POSITION_UPDATER.compareAndSet(ManagedCursorImpl.this, + mdEntry.newPosition, null); + // Remove from the individual deleted messages all the entries before the new mark delete // point. lock.writeLock().lock(); @@ -1828,11 +1892,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(); } @@ -1841,22 +1901,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); @@ -1867,13 +1918,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); } }); } From 12f03ddf1fe9f17f32127b25130db5695801dc37 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 6 Apr 2022 16:25:05 +0300 Subject: [PATCH 2/4] [ML] Resetting should reset lastMarkDeleteEntry --- .../org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 1 + 1 file changed, 1 insertion(+) 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 77b0aaeb9be1d..5a8c1a0f0dc58 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 @@ -1156,6 +1156,7 @@ 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 From 9344d9920d8f642d216fa74df1aa072b68d67176 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 12 Apr 2022 17:25:22 +0300 Subject: [PATCH 3/4] [ML] Reset fields in initializeCursorPosition method --- .../org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 3 +++ 1 file changed, 3 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 5a8c1a0f0dc58..82aae3e8effdc 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 @@ -1613,6 +1613,9 @@ boolean hasMoreEntries(PositionImpl position) { void initializeCursorPosition(Pair 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. From 1b89b52dec44882b71915b0a946d4eba111b0b8d Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 13 Apr 2022 09:24:03 +0300 Subject: [PATCH 4/4] Address review feedback: latest -> later in log message --- .../org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 82aae3e8effdc..084c31641ecc7 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 @@ -1834,7 +1834,7 @@ void internalMarkDelete(final MarkDeleteEntry mdEntry) { && mdEntry.newPosition.compareTo(persistentMarkDeletePosition) < 0) { if (log.isInfoEnabled()) { log.info("Skipping updating mark delete position to {}. The persisted mark delete position {} " - + "is latest.", mdEntry.newPosition, persistentMarkDeletePosition); + + "is later.", mdEntry.newPosition, persistentMarkDeletePosition); } mdEntry.triggerComplete(); return; @@ -1852,7 +1852,7 @@ void internalMarkDelete(final MarkDeleteEntry mdEntry) { if (inProgressLatest != mdEntry.newPosition) { if (log.isInfoEnabled()) { log.info("Skipping updating mark delete position to {}. The mark delete position update " - + "in progress {} is latest.", mdEntry.newPosition, inProgressLatest); + + "in progress {} is later.", mdEntry.newPosition, inProgressLatest); } mdEntry.triggerComplete(); return;