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 f4f695714d5c2..a5cabe2cde3b2 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 @@ -1742,7 +1742,7 @@ public void asyncMarkDelete(final Position position, Map propertie // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { isDirty = true; - lastMarkDeleteEntry = new MarkDeleteEntry(newPosition, properties, null, null); + updateLastMarkDeleteEntryToLatest(newPosition, properties); callback.markDeleteComplete(ctx); return; } @@ -1796,7 +1796,14 @@ void internalMarkDelete(final MarkDeleteEntry mdEntry) { // ledger is postponed to when the counter goes to 0. PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.incrementAndGet(this); - lastMarkDeleteEntry = mdEntry; + LAST_MARK_DELETE_ENTRY_UPDATER.updateAndGet(this, last -> { + if (last != null && last.newPosition.compareTo(mdEntry.newPosition) > 0) { + // keep the current value since it's later then the mdEntry.newPosition + return last; + } else { + return mdEntry; + } + }); persistPositionToLedger(cursorLedger, mdEntry, new VoidCallback() { @Override @@ -2063,9 +2070,7 @@ public void asyncDelete(Iterable positions, AsyncCallbacks.DeleteCallb // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { isDirty = true; - PositionImpl finalNewMarkDeletePosition = newMarkDeletePosition; - LAST_MARK_DELETE_ENTRY_UPDATER.updateAndGet(this, - last -> new MarkDeleteEntry(finalNewMarkDeletePosition, last.properties, null, null)); + updateLastMarkDeleteEntryToLatest(newMarkDeletePosition, null); callback.deleteComplete(ctx); return; } @@ -2097,6 +2102,22 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) { } } + // update lastMarkDeleteEntry field if newPosition is later than the current lastMarkDeleteEntry.newPosition + private void updateLastMarkDeleteEntryToLatest(final PositionImpl newPosition, + final Map properties) { + LAST_MARK_DELETE_ENTRY_UPDATER.updateAndGet(this, last -> { + if (last != null && last.newPosition.compareTo(newPosition) > 0) { + // keep current value, don't update + return last; + } else { + // use given properties or when missing, use the properties from the previous field value + Map propertiesToUse = + properties != null ? properties : (last != null ? last.properties : Collections.emptyMap()); + return new MarkDeleteEntry(newPosition, propertiesToUse, null, null); + } + }); + } + /** * Given a list of entries, filter out the entries that have already been individually deleted. *