[ML] Fix race condition in updating lastMarkDeleteEntry field
missed updates can lead to the subscription and consuming getting stuck
@@ -1742,7 +1742,7 @@ public class ManagedCursorImpl implements ManagedCursor {
// 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 @@ public class ManagedCursorImpl implements ManagedCursor {
// 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 class ManagedCursorImpl implements ManagedCursor {
// 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 class ManagedCursorImpl implements ManagedCursor {
}
}
+ // update lastMarkDeleteEntry field if newPosition is later than the current lastMarkDeleteEntry.newPosition
+ private void updateLastMarkDeleteEntryToLatest(final PositionImpl newPosition,
+ final Map<String, Long> 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<String, Long> 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.
*
