Fixed race condition in cursor.asyncDelete()
@@ -1173,8 +1173,6 @@ public class ManagedCursorImpl implements ManagedCursor {
* @return the previous acknowledged position
*/
PositionImpl setAcknowledgedPosition(PositionImpl newMarkDeletePosition) {
- lock.writeLock().lock();
- try {
if (newMarkDeletePosition.compareTo(markDeletePosition) < 0) {
throw new IllegalArgumentException("Mark deleting an already mark-deleted position");
}
@@ -1222,9 +1220,7 @@ public class ManagedCursorImpl implements ManagedCursor {
markDeletePosition = PositionImpl.get(newMarkDeletePosition);
individualDeletedMessages.remove(Range.atMost(markDeletePosition));
oldMarkDeletePosition.recycle();
- } finally {
- lock.writeLock().unlock();
- }
+
return newMarkDeletePosition;
}
@@ -1254,11 +1250,14 @@ public class ManagedCursorImpl implements ManagedCursor {
}
PositionImpl newPosition = (PositionImpl) position;
+ lock.writeLock().lock();
try {
newPosition = setAcknowledgedPosition(newPosition);
} catch (IllegalArgumentException e) {
callback.markDeleteFailed(new ManagedLedgerException(e), ctx);
return;
+ } finally {
+ lock.writeLock().unlock();
}
// Apply rate limiting to mark-delete operations
@@ -1487,6 +1486,12 @@ public class ManagedCursorImpl implements ManagedCursor {
newMarkDeletePosition = range.upperEndpoint();
}
}
+
+ if (newMarkDeletePosition != null) {
+ newMarkDeletePosition = setAcknowledgedPosition(newMarkDeletePosition);
+ } else {
+ newMarkDeletePosition = markDeletePosition;
+ }
} catch (Exception e) {
log.warn("[{}] [{}] Error while updating individualDeletedMessages [{}]", ledger.getName(), name,
e.getMessage(), e);
@@ -1496,19 +1501,13 @@ public class ManagedCursorImpl implements ManagedCursor {
lock.writeLock().unlock();
}
- try {
- if (newMarkDeletePosition != null) {
- newMarkDeletePosition = setAcknowledgedPosition(newMarkDeletePosition);
- } else {
- newMarkDeletePosition = markDeletePosition;
- }
-
// Apply rate limiting to mark-delete operations
if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) {
callback.deleteComplete(ctx);
return;
}
+ try {
internalAsyncMarkDelete(newMarkDeletePosition, new MarkDeleteCallback() {
@Override
public void markDeleteComplete(Object ctx) {
