You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2025/07/26 01:15:27 UTC
[PR] [WIP][fix][broker] Fix remaining ManagedCursorImpl issues around state management for pending reads [pulsar]
lhotari opened a new pull request, #24569:
URL: https://github.com/apache/pulsar/pull/24569
Fixes #24566
### Motivation
Test `ReplicatorTest.testCloseReplicatorStartProducer` was failing and investigation shared in https://github.com/apache/pulsar/issues/24566#issuecomment-3119993989 by @3pacccccc revealed behavior that was caused by #24551 changes.
### Modifications
- revert some of the #24551 changes and go back to compare-and-swap based solution due to performance concerns and correctness concerns.
- Op in the current thread could get recycled. Add id field to OpRead that is passed as a parameter to checkForNewEntries
- Change the CAS operation to use a Function with `getAndUpdate` method on WAITING_READ_OP_UPDATER, which checks the OpRead instance's id before resetting the field value.
```java
// Try to cancel the notification request
// Clear the waiting read op only if it matches the current opReadId
OpReadEntry cancelledReadOp = WAITING_READ_OP_UPDATER.getAndUpdate(this,
current -> current == op && current.id == opReadId ? null : current);
if (cancelledReadOp != null && cancelledReadOp.id == opReadId) {
```
- handle waitingReadOp cases when the cursor is closed so that the waitingReadOp gets completed with CursorAlreadyClosedException.
- prevent multiple races in closing by a dummy OpRead singleton instance `WAITING_READ_OP_FOR_CLOSED_CURSOR`.
- this is set into the `waitingReadOp` field as a tombstone value when the cursor is closing
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3126090308
@lhotari
I pushed a PR to fix the root cause
- https://github.com/apache/pulsar/pull/24573
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3127443551
> > Adding `Runnable` as the parameter of `addWaitingCursorRequested` and `removeWaitingCursor` makes code a bit hard to read. When I read the code, I was wondering how many places could it be called and found only one place. IMO, it's a over-abstraction.
@BewareMyPower Yes, you are right about that it's hard to read. The reason why I didn't directly do what you are suggesting is that it's a bad idea to mutate the internal state of another instance. That breaks object encapsulation principles.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237186004
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
Review Comment:
There's a possibility that deletion fails. In that case, the current state would be `DeletingFailed` and that that would allow this check to proceed so that deletion could be retried.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237319322
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,13 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read. This is because internally,
+ // in the dispatcher implementations, there is a "havePendingRead" flag that is not reset. If the pending
+ // read is cancelled, the dispatcher will not continue reading from the managed ledger when a new
+ // consumer is added to the dispatcher since based on the "havePendingRead" state, it will continue to
+ // expect that a read is pending and will not submit a new read.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
Comment improved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2238748448
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -235,6 +227,11 @@ public class ManagedCursorImpl implements ManagedCursor {
// active state cache in ManagedCursor. It should be in sync with the state in activeCursors in ManagedLedger.
private volatile boolean isActive = false;
+ // This is a lock used to update the registration state of the cursor in the managed ledger.
+ private final Object waitingRegistrationLock = new Object();
Review Comment:
Thanks, I'll rename this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2238749148
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -235,6 +227,11 @@ public class ManagedCursorImpl implements ManagedCursor {
// active state cache in ManagedCursor. It should be in sync with the state in activeCursors in ManagedLedger.
private volatile boolean isActive = false;
+ // This is a lock used to update the registration state of the cursor in the managed ledger.
+ private final Object waitingRegistrationLock = new Object();
+ // This is used to track if the cursor is waiting for registration in the managed ledger.
+ boolean waitingRegistered = false;
Review Comment:
Thanks, I'll rename this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234126689
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3632,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state=", ledger.getName(), name, state);
Review Comment:
The log statement has incorrect formatting. The placeholder '{}' for the state parameter is missing. It should be '[{}-{}] Cursor ledger is already deleting or deleted. state={}' to properly format the state value.
```suggestion
log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236893829
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java:
##########
@@ -34,7 +35,14 @@
import org.slf4j.LoggerFactory;
class OpReadEntry implements ReadEntriesCallback {
-
+ static final OpReadEntry WAITING_READ_OP_FOR_CLOSED_CURSOR = new OpReadEntry();
Review Comment:
Sorry, ignore this comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236828927
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1174,61 +1188,29 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
@Override
public boolean isClosed() {
- return state == State.Closed || state == State.Closing;
+ return state.isClosed();
}
@Override
public boolean cancelPendingReadRequest() {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancel pending read request", ledger.getName(), name);
}
- synchronized (pendingReadOpMutex) {
- final OpReadEntry op = WAITING_READ_OP_UPDATER.get(this);
- // Case 1: the pending read has executed, or there is no pending read.
- if (op == null) {
- return false;
- }
- Function<Boolean, Boolean> clearWaitingReadOp = removeWaitingCursor -> {
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- if (removeWaitingCursor) {
- ledger.removeWaitingCursor(this);
- }
- op.recycle();
- return true;
- } else {
- // Managed ledger has noticed that new entries was added.
- if (WAITING_READ_OP_UPDATER.get(this) == null) {
- return false;
- }
- // It will never occur, it means the lock "pendingReadOpMutex" does not work as expected, which
- // allowed other thread to modify the "waitingReadOp" concurrently.
- // The waitingReadOp has been modified to other instance, which will never occur.
- log.warn("[{}] [{}] Cancel pending request encountered an unexpected error, the lock"
- + " \"pendingReadOpMutex\" does not work as expected, which allowed other"
- + " thread to modify the \"waitingReadOp\" concurrently..", ledger.getName(), name);
- return cancelPendingReadRequest();
- }
- };
- // There is a pending read,
- // Case 2: delayCheckForNewEntriesTask can be cancelled, no need to remove cursor from "ml.waitingCursors",
- // because it has not added successfully yet.
- if (delayCheckForNewEntriesTask != null && delayCheckForNewEntriesTask.cancel()) {
- return clearWaitingReadOp.apply(false);
- }
- // Case 3: managedLedgerNewEntriesCheckDelayInMillis is "0".
- // Case 4: delayCheckForNewEntriesTask has done, which has added cursor into "ml.waitingCursors".
- if (delayCheckForNewEntriesTask == null || delayCheckForNewEntriesTask.isDone()) {
- return clearWaitingReadOp.apply(true);
+ final OpReadEntry op = WAITING_READ_OP_UPDATER.getAndUpdate(this, current -> {
+ if (current == OpReadEntry.WAITING_READ_OP_FOR_CLOSED_CURSOR) {
+ return current;
}
- // Case 5: delayCheckForNewEntriesTask is running, but not done, which only occurs at a corner case. It
- // only happens when the task is starting. Calling "cancelPendingReadRequest" here will release the lock
- // "pendingReadOpMutex" and let the task go ahead.
- return cancelPendingReadRequest();
+ return null;
+ });
+ if (op != null) {
+ op.recycle();
}
+ return op != null;
Review Comment:
It also returns true even when `op` is `WAITING_READ_OP_FOR_CLOSED_CURSOR`. However, it means `closeWaitingCursor` has been called that the cursor has been removed from `waitingCursors`. Returning true here does not make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237313693
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
Review Comment:
actually, the reason why it's relatively expensive is the scanning of the complete ConcurrentLinkedQueue in ManagedLedgerImpl to remove the instance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236859028
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
ledger.asyncReadEntries(op);
} else {
- log.info("[{}] [{}] notification that new entries added was already be called, skipped the current"
Review Comment:
Okay, it will not trigger a repeated reading, just left an orphan cursor in `ml.waitingCursors`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237921418
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,16 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read with cursor.cancelPendingReadRequest.
+ // This is because internally, in the dispatcher implementations, there is a "havePendingRead" flag
+ // that is not reset. If the pending read is cancelled, the dispatcher will not continue reading from
+ // the managed ledger when a new consumer is added to the dispatcher since based on the "havePendingRead"
+ // state, it will continue to expect that a read is pending and will not submit a new read.
+ // For non-durable cursors, there's no difference since the cursor is not expected to be used again.
+
+ // remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
If we look back at the first issue that was fixed by https://github.com/apache/pulsar/pull/24551,
```
(Highlight) The method scheduleReadOnActiveConsumer resets the variable cursor.waitingReadOp, but it forgets to remove the cursor from managedledger.waitingCursors
```
https://github.com/apache/pulsar/pull/24551/files#diff-1d721da3aaff7689077d33bf3e55fd8d6a5c9184549857a6120d8050c5d67798L385-L386
Here, try to add another one.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -235,6 +227,11 @@ public class ManagedCursorImpl implements ManagedCursor {
// active state cache in ManagedCursor. It should be in sync with the state in activeCursors in ManagedLedger.
private volatile boolean isActive = false;
+ // This is a lock used to update the registration state of the cursor in the managed ledger.
+ private final Object waitingRegistrationLock = new Object();
+ // This is used to track if the cursor is waiting for registration in the managed ledger.
+ boolean waitingRegistered = false;
Review Comment:
```suggestion
boolean registeredToWaitingCursors = false;
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -235,6 +227,11 @@ public class ManagedCursorImpl implements ManagedCursor {
// active state cache in ManagedCursor. It should be in sync with the state in activeCursors in ManagedLedger.
private volatile boolean isActive = false;
+ // This is a lock used to update the registration state of the cursor in the managed ledger.
+ private final Object waitingRegistrationLock = new Object();
Review Comment:
```suggestion
private final Object registerToWaitingCursorsLock = new Object();
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,16 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read with cursor.cancelPendingReadRequest.
+ // This is because internally, in the dispatcher implementations, there is a "havePendingRead" flag
+ // that is not reset. If the pending read is cancelled, the dispatcher will not continue reading from
+ // the managed ledger when a new consumer is added to the dispatcher since based on the "havePendingRead"
+ // state, it will continue to expect that a read is pending and will not submit a new read.
+ // For non-durable cursors, there's no difference since the cursor is not expected to be used again.
+
+ // remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
Can we optimize the method of the ManagedCursor to remove from the waiting cursors if the cursor has been registered to the managed ledger? So that we can avoid call `topic.getManagedLedger().removeWaitingCursor(cursor);` here.
```
public void setInactive() {
if (isActive) {
ledger.deactivateCursor(this);
isActive = false;
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234984518
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1040,87 +1089,36 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
-
- synchronized (pendingReadOpMutex) {
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
- op.readPosition);
- }
-
- // Check again for new entries after the configured time, then if still no entries are available
- // register to be notified.
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
- } else {
- // If there's no delay, check directly from the same thread
- checkForNewEntries(op, callback, ctx);
- }
+ int opReadId = op.id;
Review Comment:
could we check like this?
```java
OpReadEntry op = OpReadEntry.EMPTY_PLACE_HOLDER;
if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
op.recycle();
callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
return;
}
op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
int opReadId = op.id;
WAITING_READ_OP_UPDATER.set(this, op);
```
create a new EMPTY_PLACE_HOLDER just like ```OpReadEntry#WAITING_READ_OP_FOR_CLOSED_CURSOR``` which can reduce a lot of useless OpReadEntry create and recycle while thread racing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235676870
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1046,7 +1045,7 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 15000)
+ @Test(timeOut = 30000)
public void testCloseReplicatorStartProducer() throws Exception {
Review Comment:
> Since `WAITING_READ_OP_UPDATER` can only be compared and set to `null` at most once, how can it be incorrectly reused?
`cursor.cancelPendingReadRequest` or might be called in the main thread before the scheduled `checkForNewEntries` completes. `cursor.notifyEntriesAvailable` might also be called to to complete the `waitingReadOp` before `checkForNewEntries` completes. The main thread could then call `asyncReadEntriesWithSkipOrWait` and the get the same `OpReadEntry` instance from the recycler (since Netty's recycler is thread local).
When the `checkForNewEntries` completes, the `compareAndSet` check would succeed since the same instance continues to be in the `waitingReadOp` field. I guess this would result in duplicate read entries in most cases.
This instance mix-up could happen across different `managedCursorImpl` instances. Without instance id checks for handling `waitingReadOp` CAS operations, it leads to non-deterministic results as you can see from the few examples above. That's why there's a need to have a `id` for `OpReadEntry`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235766127
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1046,7 +1045,7 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 15000)
+ @Test(timeOut = 30000)
public void testCloseReplicatorStartProducer() throws Exception {
Review Comment:
@poorbarcode Not in general. When using recycled objects, it's error prone to use CAS (`compareAndSet`) method since there might be cases where the object is recycled in between and it's a different usage of the same object instance. It's possible that your changes in PR 24551 prevent that from happening by using synchronization. Since ManagedCursorImpl operations are on the performance hot path, we should avoid synchronization as much as possible and continue to use the CAS based solution with proper identity check by using `getAndUpdate` where the existing value could be checked before making the update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237233003
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -791,7 +808,39 @@ private void recoveredCursor(Position position, Map<String, Long> properties,
// assign cursor-ledger so, it can be deleted when new ledger will be switched
this.cursorLedger = recoveredFromCursorLedger;
this.isCursorLedgerReadOnly = true;
- STATE_UPDATER.set(this, State.NoLedger);
+ changeStateIfNotClosed(State.NoLedger);
+ }
+
+ /**
+ * Change the state of the cursor if it is not already considered closed.
+ * This is to prevent invalid state transitions when the cursor is already closed.
+ *
+ * @param newState The new state to set
+ * @return The previous state of the cursor
+ */
+ private State changeStateIfNotClosed(State newState) {
+ return STATE_UPDATER.getAndUpdate(this, current -> {
+ if (current.isClosed()) {
+ return current;
+ }
+ return newState;
+ });
+ }
+
+ /**
+ * Change the state of the cursor if it is not already Deleting or Deleted.
+ * This is to prevent invalid state transitions when the cursor is already being deleted or has been deleted.
+ *
+ * @param newState The new state to set
+ * @return The previous state of the cursor
+ */
+ State changeStateIfNotDeletingOrDeleted(State newState) {
Review Comment:
True. I'll inline this method since it's only used in one location. It just adds confusion to have this method around.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235312969
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1040,87 +1089,36 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
-
- synchronized (pendingReadOpMutex) {
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
- op.readPosition);
- }
-
- // Check again for new entries after the configured time, then if still no entries are available
- // register to be notified.
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
- } else {
- // If there's no delay, check directly from the same thread
- checkForNewEntries(op, callback, ctx);
- }
+ int opReadId = op.id;
Review Comment:
I mean , We can optimize the code by avoiding unnecessary OpReadEntry.create() and OpReadEntry.recycle() operations when WAITING_READ_OP_UPDATER is not null. This improvement offers two key benefits:
- Resource Efficiency - Eliminates wasteful object creation and recycling when the operation would immediately fail
- ID Management - Since ```OpReadEntry.id``` auto-increments in this case, this may prevents ```OpReadEntry.id``` grow
rapidly and only few ```OpReadEntry.id``` are actually used for valid operations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236795670
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -4012,4 +4034,55 @@ public ManagedLedgerInternalStats.CursorStats getCursorStats() {
cs.properties = getProperties();
return cs;
}
+
+ /**
+ * Called by ManagedLedgerImpl to execute the Runnable inside the lock to remove the cursor from it's
+ * waiting cursors list.
+ * The cursor state is set to unregistered, and it can be registered again for waiting in ManagedLedgerImpl.
+ */
+ void removeWaitingCursorRequested(Runnable removeWaitingCursorRunnable) {
+ synchronized (waitingRegistrationLock) {
Review Comment:
to optimize, we could remove `ledger.removeWaitingCursor(this)` in `ManagedCursorImpl` completely for the read cycle since the waiting cursor will get removed when the entries are notified or when ledger.removeWaitingCursor is called from other parts of the code (`PersistentSubscription`). As mentioned in https://github.com/apache/pulsar/pull/24569#discussion_r2235714944, the removal from the `ConcurrentLinkedQueue` is potentially costly since it has to iterate through the complete queue to remove a cursor. That's why it would be better to avoid that operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235479127
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1046,7 +1045,7 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 15000)
+ @Test(timeOut = 30000)
public void testCloseReplicatorStartProducer() throws Exception {
Review Comment:
Hi @lhotari
> Multiple threads attempt to cancel pending read requests concurrently
> Cursors are closed while read operations are pending
> OpReadEntry instances are recycled and reused incorrectly
`cursor.cancelPendingReadRequest` will compare and set `WAITING_READ_OP_UPDATER` to `null`, then recycle ` OpReadEntry`.
Since `WAITING_READ_OP_UPDATER` can only be compared and set to `null` at most once, how can it be incorrectly reused?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237318231
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -4012,4 +4034,55 @@ public ManagedLedgerInternalStats.CursorStats getCursorStats() {
cs.properties = getProperties();
return cs;
}
+
+ /**
+ * Called by ManagedLedgerImpl to execute the Runnable inside the lock to remove the cursor from it's
+ * waiting cursors list.
+ * The cursor state is set to unregistered, and it can be registered again for waiting in ManagedLedgerImpl.
+ */
+ void removeWaitingCursorRequested(Runnable removeWaitingCursorRunnable) {
+ synchronized (waitingRegistrationLock) {
Review Comment:
There's no calls to `ledger.removeWaitingCursor(this)` in the hot path, so this should be fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3128209129
> I've gone through the whole PR, overall LGTM except for some coding style issues and some questions I've left. Please address these comments before merging.
>
> P.S. The state machine is hard to read. It would be helpful to have a graph to show all possible state changes, but it's not required.
Thank you @BewareMyPower. I've addressed your review comments. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2238711971
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,32 +3622,62 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State beforeChangingState = changeStateToDeletingIfNotDeleted();
+ if (beforeChangingState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleted. state={}", ledger.getName(), name,
+ beforeChangingState);
Review Comment:
Correct. Code might change in the future, so having the logging in place is useful from that perspective.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234139522
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
return;
}
ledger.mbean.startCursorLedgerDeleteOp();
bookkeeper.asyncDeleteLedger(cursorLedger.getId(), (rc, ctx) -> {
ledger.mbean.endCursorLedgerDeleteOp();
if (rc == BKException.Code.OK) {
+ state = State.Deleted;
Review Comment:
considered
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
Review Comment:
considered
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235741306
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
ledger.asyncReadEntries(op);
} else {
- log.info("[{}] [{}] notification that new entries added was already be called, skipped the current"
Review Comment:
In this case, `opReadEntry` in `waitingReadOp` was null or some other instance so it was either cancelled or notified before `checkNewEntries` was called.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237173483
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1174,61 +1188,29 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
@Override
public boolean isClosed() {
- return state == State.Closed || state == State.Closing;
+ return state.isClosed();
}
@Override
public boolean cancelPendingReadRequest() {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancel pending read request", ledger.getName(), name);
}
- synchronized (pendingReadOpMutex) {
- final OpReadEntry op = WAITING_READ_OP_UPDATER.get(this);
- // Case 1: the pending read has executed, or there is no pending read.
- if (op == null) {
- return false;
- }
- Function<Boolean, Boolean> clearWaitingReadOp = removeWaitingCursor -> {
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- if (removeWaitingCursor) {
- ledger.removeWaitingCursor(this);
- }
- op.recycle();
- return true;
- } else {
- // Managed ledger has noticed that new entries was added.
- if (WAITING_READ_OP_UPDATER.get(this) == null) {
- return false;
- }
- // It will never occur, it means the lock "pendingReadOpMutex" does not work as expected, which
- // allowed other thread to modify the "waitingReadOp" concurrently.
- // The waitingReadOp has been modified to other instance, which will never occur.
- log.warn("[{}] [{}] Cancel pending request encountered an unexpected error, the lock"
- + " \"pendingReadOpMutex\" does not work as expected, which allowed other"
- + " thread to modify the \"waitingReadOp\" concurrently..", ledger.getName(), name);
- return cancelPendingReadRequest();
- }
- };
- // There is a pending read,
- // Case 2: delayCheckForNewEntriesTask can be cancelled, no need to remove cursor from "ml.waitingCursors",
- // because it has not added successfully yet.
- if (delayCheckForNewEntriesTask != null && delayCheckForNewEntriesTask.cancel()) {
- return clearWaitingReadOp.apply(false);
- }
- // Case 3: managedLedgerNewEntriesCheckDelayInMillis is "0".
- // Case 4: delayCheckForNewEntriesTask has done, which has added cursor into "ml.waitingCursors".
- if (delayCheckForNewEntriesTask == null || delayCheckForNewEntriesTask.isDone()) {
- return clearWaitingReadOp.apply(true);
+ final OpReadEntry op = WAITING_READ_OP_UPDATER.getAndUpdate(this, current -> {
+ if (current == OpReadEntry.WAITING_READ_OP_FOR_CLOSED_CURSOR) {
+ return current;
}
- // Case 5: delayCheckForNewEntriesTask is running, but not done, which only occurs at a corner case. It
- // only happens when the task is starting. Calling "cancelPendingReadRequest" here will release the lock
- // "pendingReadOpMutex" and let the task go ahead.
- return cancelPendingReadRequest();
+ return null;
+ });
+ if (op != null) {
+ op.recycle();
}
+ return op != null;
Review Comment:
I'll change to `op != null && op !=
OpReadEntry.WAITING_READ_OP_FOR_CLOSED_CURSOR`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236862784
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
Review Comment:
The log seems wrong. In `changeStateIfNotDeletingOrDeleted`, there are only three cases:
| previousState | state |
| :- | :- |
| Deleting | Deleting |
| Deleted | Deleted |
| Other | Deleting |
So `state` should always be `Deleted` here. The log `is already deleting or deleted` should be `is already deleted`
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -791,7 +808,39 @@ private void recoveredCursor(Position position, Map<String, Long> properties,
// assign cursor-ledger so, it can be deleted when new ledger will be switched
this.cursorLedger = recoveredFromCursorLedger;
this.isCursorLedgerReadOnly = true;
- STATE_UPDATER.set(this, State.NoLedger);
+ changeStateIfNotClosed(State.NoLedger);
+ }
+
+ /**
+ * Change the state of the cursor if it is not already considered closed.
+ * This is to prevent invalid state transitions when the cursor is already closed.
+ *
+ * @param newState The new state to set
+ * @return The previous state of the cursor
+ */
+ private State changeStateIfNotClosed(State newState) {
+ return STATE_UPDATER.getAndUpdate(this, current -> {
+ if (current.isClosed()) {
+ return current;
+ }
+ return newState;
+ });
+ }
+
+ /**
+ * Change the state of the cursor if it is not already Deleting or Deleted.
+ * This is to prevent invalid state transitions when the cursor is already being deleted or has been deleted.
+ *
+ * @param newState The new state to set
+ * @return The previous state of the cursor
+ */
+ State changeStateIfNotDeletingOrDeleted(State newState) {
Review Comment:
`newState` is always `Deleting`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235462754
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1046,7 +1045,7 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 15000)
+ @Test(timeOut = 30000)
public void testCloseReplicatorStartProducer() throws Exception {
Review Comment:
Start a channel to discuss the following topic
@poorbarcode left a comment
> Multiple threads attempt to cancel pending read requests concurrently
> Cursors are closed while read operations are pending
> OpReadEntry instances are recycled and reused incorrectly
Could you describe the details of how the `OpReadEntry` will be reused
---
@lhotari left a comment
Since OpReadEntry instances are recycled, they are reused. It's not sufficient do use compareAndSet. There's a need to have some unique identifier on the instance to do the CAS operation comparison. It's possible to find code paths where this could happen and new problems could be introduced unless we have a proper solution to for CAS operations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2238717165
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,16 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read with cursor.cancelPendingReadRequest.
+ // This is because internally, in the dispatcher implementations, there is a "havePendingRead" flag
+ // that is not reset. If the pending read is cancelled, the dispatcher will not continue reading from
+ // the managed ledger when a new consumer is added to the dispatcher since based on the "havePendingRead"
+ // state, it will continue to expect that a read is pending and will not submit a new read.
+ // For non-durable cursors, there's no difference since the cursor is not expected to be used again.
+
+ // remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
> Can we optimize the method of the ManagedCursor to remove from the waiting cursors if the cursor has been registered to the managed ledger? So that we can avoid call `topic.getManagedLedger().removeWaitingCursor(cursor);` here.
>
> ```
> public void setInactive() {
> if (isActive) {
> ledger.deactivateCursor(this);
> isActive = false;
> }
> }
> ```
deactivateCursor doesn't remove the cursor from waiting cursors and it shouldn't. I experimented with that and found out that it would also cause a state inconsistency with dispatcher's "havePendingRead" and make dispatchers to stop receiving messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235714944
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
Review Comment:
That's true that it could be done, however, it doesn't seem to be necessary in this solution. Calling `removeWaitingCursor` is potentially an expensive operation since `waitingCursors` is a `ConcurrentLinkedQueue` in `ManagedLedgerImpl` and when there would be a large amount of waiting cursors. That's why I'd like to avoid calling `removeWaitingCursor` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2238350271
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,32 +3622,62 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State beforeChangingState = changeStateToDeletingIfNotDeleted();
+ if (beforeChangingState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleted. state={}", ledger.getName(), name,
+ beforeChangingState);
Review Comment:
Logging the state seems meaningless because `beforeChangingState ` is always `Deleted`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236916581
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,13 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read. This is because internally,
+ // in the dispatcher implementations, there is a "havePendingRead" flag that is not reset. If the pending
+ // read is cancelled, the dispatcher will not continue reading from the managed ledger when a new
+ // consumer is added to the dispatcher since based on the "havePendingRead" state, it will continue to
+ // expect that a read is pending and will not submit a new read.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
It's still hard to understand the motivation here even with the comments. The comments also make me confused.
> For durable cursors
The code here does not check if the cursor is durable.
> cancel the pending read
But it does not call `cursor.cancelPendingReadRequest()`.
Is it possible to add a test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236741844
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -4012,4 +4034,55 @@ public ManagedLedgerInternalStats.CursorStats getCursorStats() {
cs.properties = getProperties();
return cs;
}
+
+ /**
+ * Called by ManagedLedgerImpl to execute the Runnable inside the lock to remove the cursor from it's
+ * waiting cursors list.
+ * The cursor state is set to unregistered, and it can be registered again for waiting in ManagedLedgerImpl.
+ */
+ void removeWaitingCursorRequested(Runnable removeWaitingCursorRunnable) {
+ synchronized (waitingRegistrationLock) {
Review Comment:
This method will be called on the publishing messages path, which is a really hot path, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #24569:
URL: https://github.com/apache/pulsar/pull/24569
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2240329579
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,16 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read with cursor.cancelPendingReadRequest.
+ // This is because internally, in the dispatcher implementations, there is a "havePendingRead" flag
+ // that is not reset. If the pending read is cancelled, the dispatcher will not continue reading from
+ // the managed ledger when a new consumer is added to the dispatcher since based on the "havePendingRead"
+ // state, it will continue to expect that a read is pending and will not submit a new read.
+ // For non-durable cursors, there's no difference since the cursor is not expected to be used again.
+
+ // remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
> My concern is that we used some internal concepts from the ManagedLedger in PersistentSubscription. We should find a way to avoid that. The state of `waitingCursors` should be self-managed by ManagedLedger.
I completely agree that it should be avoided. There should have been some sort of event for handling this.
btw. The call `topic.getManagedLedger().removeWaitingCursor(cursor)` wasn't added in this PR. Before #24551, there was a similar call. The call `topic.getManagedLedger().removeWaitingCursor(cursor)` was added to `PersistentSubscription` in #13939. Another PR in this area was #22454.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2240302991
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,16 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read with cursor.cancelPendingReadRequest.
+ // This is because internally, in the dispatcher implementations, there is a "havePendingRead" flag
+ // that is not reset. If the pending read is cancelled, the dispatcher will not continue reading from
+ // the managed ledger when a new consumer is added to the dispatcher since based on the "havePendingRead"
+ // state, it will continue to expect that a read is pending and will not submit a new read.
+ // For non-durable cursors, there's no difference since the cursor is not expected to be used again.
+
+ // remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
My concern is that we used some internal concepts from the ManagedLedger in PersistentSubscription. We should find a way to avoid that. The state of `waitingCursors` should be self-managed by ManagedLedger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234138255
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java:
##########
@@ -208,6 +233,11 @@ protected OpReadEntry newObject(Recycler.Handle<OpReadEntry> recyclerHandle) {
};
public void recycle() {
+ if (recyclerHandle == null) {
+ // This is the no-op instance, do not recycle
+ return;
+ }
+ id = -1;
Review Comment:
[nitpick] Setting id to -1 in recycle() could cause confusion since negative values are explicitly mentioned as valid due to integer overflow. Consider using a more explicit sentinel value or documenting why -1 is chosen.
```suggestion
id = SENTINEL_ID;
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
return;
}
ledger.mbean.startCursorLedgerDeleteOp();
bookkeeper.asyncDeleteLedger(cursorLedger.getId(), (rc, ctx) -> {
ledger.mbean.endCursorLedgerDeleteOp();
if (rc == BKException.Code.OK) {
+ state = State.Deleted;
log.info("[{}][{}] Deleted cursor ledger {}", ledger.getName(), name, cursorLedger.getId());
} else {
log.warn("[{}][{}] Failed to delete ledger {}: {}", ledger.getName(), name, cursorLedger.getId(),
BKException.getMessage(rc));
if (!isNoSuchLedgerExistsException(rc)) {
+ state = State.DeletingFailed;
ledger.getScheduledExecutor().schedule(() -> asyncDeleteCursorLedger(retry - 1),
DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
+ } else {
+ state = State.Deleted;
Review Comment:
Direct state assignment bypasses atomic operations used elsewhere in the class. Consider using STATE_UPDATER.set() for consistency with the atomic state management pattern.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -235,6 +227,11 @@ public class ManagedCursorImpl implements ManagedCursor {
// active state cache in ManagedCursor. It should be in sync with the state in activeCursors in ManagedLedger.
private volatile boolean isActive = false;
+ // This is a lock used to update the registration state of the cursor in the managed ledger.
+ private Object waitingRegistrationLock = new Object();
Review Comment:
Consider making waitingRegistrationLock final since it's never reassigned after initialization.
```suggestion
private final Object waitingRegistrationLock = new Object();
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2958,12 +2940,18 @@ public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object
callback.closeComplete(ctx);
return;
}
+ closeWaitingCursor();
+ setInactive();
persistPositionWhenClosing(lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties,
new AsyncCallbacks.CloseCallback(){
@Override
public void closeComplete(Object ctx) {
- STATE_UPDATER.set(ManagedCursorImpl.this, State.Closed);
+ if (!STATE_UPDATER.compareAndSet(ManagedCursorImpl.this, State.Closing, State.Closed)) {
+ log.warn("[{}] [{}] State was modified from closing to {} while closing", ledger.getName(),
+ name, state);
+ state = State.Closed;
Review Comment:
The fallback to direct assignment 'state = State.Closed' after a failed CAS operation bypasses the atomic guarantees. Consider using a retry loop with STATE_UPDATER or document why direct assignment is safe in this context.
```suggestion
while (!STATE_UPDATER.compareAndSet(ManagedCursorImpl.this, State.Closing, State.Closed)) {
log.warn("[{}] [{}] State was modified from closing to {} while closing. Retrying...",
ledger.getName(), name, state);
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
return;
}
ledger.mbean.startCursorLedgerDeleteOp();
bookkeeper.asyncDeleteLedger(cursorLedger.getId(), (rc, ctx) -> {
ledger.mbean.endCursorLedgerDeleteOp();
if (rc == BKException.Code.OK) {
+ state = State.Deleted;
log.info("[{}][{}] Deleted cursor ledger {}", ledger.getName(), name, cursorLedger.getId());
} else {
log.warn("[{}][{}] Failed to delete ledger {}: {}", ledger.getName(), name, cursorLedger.getId(),
BKException.getMessage(rc));
if (!isNoSuchLedgerExistsException(rc)) {
+ state = State.DeletingFailed;
ledger.getScheduledExecutor().schedule(() -> asyncDeleteCursorLedger(retry - 1),
DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
+ } else {
+ state = State.Deleted;
Review Comment:
Direct state assignment bypasses atomic operations used elsewhere in the class. Consider using STATE_UPDATER.set() for consistency with the atomic state management pattern.
```suggestion
STATE_UPDATER.set(this, State.Deleted);
log.info("[{}][{}] Deleted cursor ledger {}", ledger.getName(), name, cursorLedger.getId());
} else {
log.warn("[{}][{}] Failed to delete ledger {}: {}", ledger.getName(), name, cursorLedger.getId(),
BKException.getMessage(rc));
if (!isNoSuchLedgerExistsException(rc)) {
STATE_UPDATER.set(this, State.DeletingFailed);
ledger.getScheduledExecutor().schedule(() -> asyncDeleteCursorLedger(retry - 1),
DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
} else {
STATE_UPDATER.set(this, State.Deleted);
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
Review Comment:
Direct state assignment bypasses atomic operations used elsewhere in the class. Consider using STATE_UPDATER.set() for consistency with the atomic state management pattern.
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
return;
}
ledger.mbean.startCursorLedgerDeleteOp();
bookkeeper.asyncDeleteLedger(cursorLedger.getId(), (rc, ctx) -> {
ledger.mbean.endCursorLedgerDeleteOp();
if (rc == BKException.Code.OK) {
+ state = State.Deleted;
Review Comment:
Direct state assignment bypasses atomic operations used elsewhere in the class. Consider using STATE_UPDATER.set() for consistency with the atomic state management pattern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234139449
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
return;
}
ledger.mbean.startCursorLedgerDeleteOp();
bookkeeper.asyncDeleteLedger(cursorLedger.getId(), (rc, ctx) -> {
ledger.mbean.endCursorLedgerDeleteOp();
if (rc == BKException.Code.OK) {
+ state = State.Deleted;
log.info("[{}][{}] Deleted cursor ledger {}", ledger.getName(), name, cursorLedger.getId());
} else {
log.warn("[{}][{}] Failed to delete ledger {}: {}", ledger.getName(), name, cursorLedger.getId(),
BKException.getMessage(rc));
if (!isNoSuchLedgerExistsException(rc)) {
+ state = State.DeletingFailed;
ledger.getScheduledExecutor().schedule(() -> asyncDeleteCursorLedger(retry - 1),
DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
+ } else {
+ state = State.Deleted;
Review Comment:
considered
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235194161
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1040,87 +1089,36 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
-
- synchronized (pendingReadOpMutex) {
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
- op.readPosition);
- }
-
- // Check again for new entries after the configured time, then if still no entries are available
- // register to be notified.
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
- } else {
- // If there's no delay, check directly from the same thread
- checkForNewEntries(op, callback, ctx);
- }
+ int opReadId = op.id;
Review Comment:
Not really, since the point of the CAS operation is to do it atomically. OpReadEntries are recycled, so there's no useless creation of OpReadEntries. It's only about assigning fields to the instance that is from the recycling pool. If there's a race, it would be recycled immediately. The role of `WAITING_READ_OP_FOR_CLOSED_CURSOR` is different. It's a sentinel value that prevents any races to change the value when the cursor has already been closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235202509
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2246,7 +2228,7 @@ protected void internalAsyncMarkDelete(final Position newPosition, Map<String, L
// We cannot write to the ledger during the switch, need to wait until the new metadata ledger is available
synchronized (pendingMarkDeleteOps) {
// The state might have changed while we were waiting on the queue mutex
- switch (STATE_UPDATER.get(this)) {
+ switch (state) {
Review Comment:
`STATE_UPDATER.get` and `STATE_UPDATER.set` are not necessary. The volatile field can be read and set directly without `STATE_UPDATER`. The intention of the change is to use `STATE_UPDATER` only for CAS changes of the `state` field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235599193
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
Review Comment:
We need to call `removeWaitingCursor` here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235705206
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
ledger.asyncReadEntries(op);
} else {
- log.info("[{}] [{}] notification that new entries added was already be called, skipped the current"
Review Comment:
The method `cancelPendingRead` can not cancel this task, right? which causes a orphan reading more entries
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3127431821
> Adding `Runnable` as the parameter of `addWaitingCursorRequested` and `removeWaitingCursor` makes code a bit hard to read. When I read the code, I was wondering how many places could it be called and found only one place. IMO, it's a over-abstraction.
Oh yes, you are right about that. I'll fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234962829
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2246,7 +2228,7 @@ protected void internalAsyncMarkDelete(final Position newPosition, Map<String, L
// We cannot write to the ledger during the switch, need to wait until the new metadata ledger is available
synchronized (pendingMarkDeleteOps) {
// The state might have changed while we were waiting on the queue mutex
- switch (STATE_UPDATER.get(this)) {
+ switch (state) {
Review Comment:
why changing this?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1040,87 +1089,36 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
-
- synchronized (pendingReadOpMutex) {
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
- op.readPosition);
- }
-
- // Check again for new entries after the configured time, then if still no entries are available
- // register to be notified.
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
- } else {
- // If there's no delay, check directly from the same thread
- checkForNewEntries(op, callback, ctx);
- }
+ int opReadId = op.id;
Review Comment:
could we check like this?
```java
OpReadEntry op = OpReadEntry.EMPTY_PLACE_HOLDER;
int opReadId = op.id;
if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, OpReadEntry.EMPTY_PLACE_HOLDER)) {
op.recycle();
callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
return;
}
op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
```
create a new EMPTY_PLACE_HOLDER just like ```OpReadEntry#WAITING_READ_OP_FOR_CLOSED_CURSOR``` which can reduce a lot of useless OpReadEntry create while thread racing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3126240613
> > Multiple threads attempt to cancel pending read requests concurrently
> > Cursors are closed while read operations are pending
> > OpReadEntry instances are recycled and reused incorrectly
>
> Could you describe the detauls of how the `OpReadEntry` will be reused
@poorbarcode Since OpReadEntry instances are recycled, they are reused. It's not sufficient do use `compareAndSet`. There's a need to have some unique identifier on the instance to do the CAS operation comparison. It's possible to find code paths where this could happen and new problems could be introduced unless we have a proper solution to for CAS operations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235676870
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1046,7 +1045,7 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 15000)
+ @Test(timeOut = 30000)
public void testCloseReplicatorStartProducer() throws Exception {
Review Comment:
> Since `WAITING_READ_OP_UPDATER` can only be compared and set to `null` at most once, how can it be incorrectly reused?
`cursor.cancelPendingReadRequest` or might be called in the main thread before the scheduled `checkForNewEntries` completes. `cursor.notifyEntriesAvailable` might also be called to to complete the `waitingReadOp` before `checkForNewEntries` completes. The main thread could then call `asyncReadEntriesWithSkipOrWait` and the get the same `OpReadEntry` instance (this is possible, but rare) from the recycler (since Netty's recycler is thread local).
When the `checkForNewEntries` completes, the `compareAndSet` check would succeed since the same instance continues to be in the `waitingReadOp` field. I guess this would result in duplicate read entries in most cases.
This instance mix-up could happen across different `managedCursorImpl` instances. Without instance id checks for handling `waitingReadOp` CAS operations, it leads to non-deterministic results as you can see from the few examples above. That's why there's a need to have a `id` for `OpReadEntry`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237286273
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -791,7 +808,39 @@ private void recoveredCursor(Position position, Map<String, Long> properties,
// assign cursor-ledger so, it can be deleted when new ledger will be switched
this.cursorLedger = recoveredFromCursorLedger;
this.isCursorLedgerReadOnly = true;
- STATE_UPDATER.set(this, State.NoLedger);
+ changeStateIfNotClosed(State.NoLedger);
+ }
+
+ /**
+ * Change the state of the cursor if it is not already considered closed.
+ * This is to prevent invalid state transitions when the cursor is already closed.
+ *
+ * @param newState The new state to set
+ * @return The previous state of the cursor
+ */
+ private State changeStateIfNotClosed(State newState) {
+ return STATE_UPDATER.getAndUpdate(this, current -> {
+ if (current.isClosed()) {
+ return current;
+ }
+ return newState;
+ });
+ }
+
+ /**
+ * Change the state of the cursor if it is not already Deleting or Deleted.
+ * This is to prevent invalid state transitions when the cursor is already being deleted or has been deleted.
+ *
+ * @param newState The new state to set
+ * @return The previous state of the cursor
+ */
+ State changeStateIfNotDeletingOrDeleted(State newState) {
Review Comment:
Actually there's 2 locations where it's used. In ManagedLedgerImpl.asyncDeleteCursor it's also used. The "Deleting" state cover all stages of deletion. If any stage fails, the state will be set to DeletingFailed, so that a retry could happen.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234984518
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1040,87 +1089,36 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
-
- synchronized (pendingReadOpMutex) {
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
- op.readPosition);
- }
-
- // Check again for new entries after the configured time, then if still no entries are available
- // register to be notified.
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
- } else {
- // If there's no delay, check directly from the same thread
- checkForNewEntries(op, callback, ctx);
- }
+ int opReadId = op.id;
Review Comment:
could we check like this?
```java
OpReadEntry op = OpReadEntry.EMPTY_PLACE_HOLDER;
if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
op.recycle();
callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
return;
}
op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
int opReadId = op.id;
WAITING_READ_OP_UPDATER.set(this, op);
```
create a new EMPTY_PLACE_HOLDER just like ```OpReadEntry#WAITING_READ_OP_FOR_CLOSED_CURSOR``` which can reduce a lot of useless OpReadEntry create while thread racing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236707468
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java:
##########
@@ -34,7 +35,14 @@
import org.slf4j.LoggerFactory;
class OpReadEntry implements ReadEntriesCallback {
-
+ static final OpReadEntry WAITING_READ_OP_FOR_CLOSED_CURSOR = new OpReadEntry();
Review Comment:
It can be a private static final field of `ManagedCursorImpl`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236855802
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
Review Comment:
Okay, `removeWaitingCursor` is expensive because it uses the `synchronized` lock
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2236994270
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1149,17 +1147,33 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
log.debug("[{}] [{}] Found more entries", ledger.getName(), name);
}
// Try to cancel the notification request
- if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
- ledger.removeWaitingCursor(this);
+ // Clear the waiting read op only if it matches the current instance and the id matches
+ // the opReadId parameter. This avoids recycled OpReadEntry instances from matching since their
+ // ids would be different after recycling.
+ OpReadEntry waitingReadOpItem = WAITING_READ_OP_UPDATER.getAndUpdate(this,
+ current -> {
+ if (current == op && current.id == opReadId) {
+ // update the value to null to cancel the waiting read op
+ return null;
+ } else {
+ // keep the current waiting read op value
+ return current;
+ }
+ });
+ // If the waiting read op was the same as the one we are trying to cancel, it means that it was now
+ // cleared from the waitingReadOp field and therefore "cancelled"
+ if (waitingReadOpItem == op && waitingReadOpItem.id == opReadId) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
ledger.asyncReadEntries(op);
} else {
- log.info("[{}] [{}] notification that new entries added was already be called, skipped the current"
Review Comment:
Correct. That would get removed when the `ml.notifyCursors` method has been called.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234139385
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
+ return;
+ }
- if (cursorLedger == null || retry <= 0) {
- if (cursorLedger != null) {
- log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
- cursorLedger.getId());
- }
+ closeWaitingCursor();
+
+ if (cursorLedger == null) {
+ return;
+ }
+
+ if (retry <= 0) {
+ log.warn("[{}-{}] Failed to delete ledger after retries {}", ledger.getName(), name,
+ cursorLedger.getId());
+ state = State.DeletingFailed;
return;
}
ledger.mbean.startCursorLedgerDeleteOp();
bookkeeper.asyncDeleteLedger(cursorLedger.getId(), (rc, ctx) -> {
ledger.mbean.endCursorLedgerDeleteOp();
if (rc == BKException.Code.OK) {
+ state = State.Deleted;
log.info("[{}][{}] Deleted cursor ledger {}", ledger.getName(), name, cursorLedger.getId());
} else {
log.warn("[{}][{}] Failed to delete ledger {}: {}", ledger.getName(), name, cursorLedger.getId(),
BKException.getMessage(rc));
if (!isNoSuchLedgerExistsException(rc)) {
+ state = State.DeletingFailed;
ledger.getScheduledExecutor().schedule(() -> asyncDeleteCursorLedger(retry - 1),
DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
+ } else {
+ state = State.Deleted;
Review Comment:
sure, valid point, but STATE_UPDATE.get and get aren't providing any value over direct referencing and assignment of a volatile field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234139007
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -2958,12 +2940,18 @@ public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object
callback.closeComplete(ctx);
return;
}
+ closeWaitingCursor();
+ setInactive();
persistPositionWhenClosing(lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties,
new AsyncCallbacks.CloseCallback(){
@Override
public void closeComplete(Object ctx) {
- STATE_UPDATER.set(ManagedCursorImpl.this, State.Closed);
+ if (!STATE_UPDATER.compareAndSet(ManagedCursorImpl.this, State.Closing, State.Closed)) {
+ log.warn("[{}] [{}] State was modified from closing to {} while closing", ledger.getName(),
+ name, state);
+ state = State.Closed;
Review Comment:
It's safe here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2234984518
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -1040,87 +1089,36 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
-
- synchronized (pendingReadOpMutex) {
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
- op.readPosition);
- }
-
- // Check again for new entries after the configured time, then if still no entries are available
- // register to be notified.
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
- } else {
- // If there's no delay, check directly from the same thread
- checkForNewEntries(op, callback, ctx);
- }
+ int opReadId = op.id;
Review Comment:
could we check like this?
```java
OpReadEntry op = OpReadEntry.EMPTY_PLACE_HOLDER;
int opReadId = op.id;
if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, OpReadEntry.EMPTY_PLACE_HOLDER)) {
op.recycle();
callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
return;
}
op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
WAITING_READ_OP_UPDATER.set(this, op);
```
create a new EMPTY_PLACE_HOLDER just like ```OpReadEntry#WAITING_READ_OP_FOR_CLOSED_CURSOR``` which can reduce a lot of useless OpReadEntry create while thread racing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3122193776
I learned through some failing tests that there are some really hairy details of the `havePendingRead` state in dispatcher and the related state in ManagedCursorImpl (`waitingReadOp`) and ManagedLedgerImpl (`waitingCursors`). It's like there's a circular runtime dependency of state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3124176264
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24569?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `69.44444%` with `55 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.30%. Comparing base ([`bbc6224`](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9d86d25`](https://app.codecov.io/gh/apache/pulsar/commit/9d86d25f1437dc146363f480bbb3163dd2a94727?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1230 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24569?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...che/bookkeeper/mledger/impl/ManagedCursorImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FManagedCursorImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRDdXJzb3JJbXBsLmphdmE=) | 67.85% | [25 Missing and 20 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FManagedLedgerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRMZWRnZXJJbXBsLmphdmE=) | 73.68% | [4 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rg/apache/bookkeeper/mledger/impl/OpReadEntry.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FOpReadEntry.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL09wUmVhZEVudHJ5LmphdmE=) | 76.92% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...che/bookkeeper/mledger/ManagedLedgerException.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2FManagedLedgerException.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9NYW5hZ2VkTGVkZ2VyRXhjZXB0aW9uLmphdmE=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24569 +/- ##
============================================
+ Coverage 73.57% 74.30% +0.73%
+ Complexity 32624 2430 -30194
============================================
Files 1877 1876 -1
Lines 139502 146391 +6889
Branches 15299 16791 +1492
============================================
+ Hits 102638 108781 +6143
- Misses 28908 28973 +65
- Partials 7956 8637 +681
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24569/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24569/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.74% <47.22%> (+2.15%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24569/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.37% <57.22%> (-0.96%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24569/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.79% <69.44%> (+0.94%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24569?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [.../bookkeeper/mledger/impl/NonDurableCursorImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FNonDurableCursorImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL05vbkR1cmFibGVDdXJzb3JJbXBsLmphdmE=) | `82.00% <100.00%> (+0.75%)` | :arrow_up: |
| [...he/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FReadOnlyCursorImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL1JlYWRPbmx5Q3Vyc29ySW1wbC5qYXZh) | `95.45% <100.00%> (+0.45%)` | :arrow_up: |
| [...ker/service/persistent/PersistentSubscription.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fpersistent%2FPersistentSubscription.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvUGVyc2lzdGVudFN1YnNjcmlwdGlvbi5qYXZh) | `77.47% <100.00%> (+0.77%)` | :arrow_up: |
| [...che/bookkeeper/mledger/ManagedLedgerException.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2FManagedLedgerException.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9NYW5hZ2VkTGVkZ2VyRXhjZXB0aW9uLmphdmE=) | `80.00% <0.00%> (-2.36%)` | :arrow_down: |
| [...rg/apache/bookkeeper/mledger/impl/OpReadEntry.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FOpReadEntry.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL09wUmVhZEVudHJ5LmphdmE=) | `75.51% <76.92%> (-8.81%)` | :arrow_down: |
| [...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FManagedLedgerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRMZWRnZXJJbXBsLmphdmE=) | `81.15% <73.68%> (+0.48%)` | :arrow_up: |
| [...che/bookkeeper/mledger/impl/ManagedCursorImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24569?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FManagedCursorImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL01hbmFnZWRDdXJzb3JJbXBsLmphdmE=) | `78.54% <67.85%> (-0.76%)` | :arrow_down: |
... and [1110 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24569/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#issuecomment-3125836280
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2238735283
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,16 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read with cursor.cancelPendingReadRequest.
+ // This is because internally, in the dispatcher implementations, there is a "havePendingRead" flag
+ // that is not reset. If the pending read is cancelled, the dispatcher will not continue reading from
+ // the managed ledger when a new consumer is added to the dispatcher since based on the "havePendingRead"
+ // state, it will continue to expect that a read is pending and will not submit a new read.
+ // For non-durable cursors, there's no difference since the cursor is not expected to be used again.
+
+ // remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
> If we look back at the first issue that was fixed by #24551,
>
> ```
> (Highlight) The method scheduleReadOnActiveConsumer resets the variable cursor.waitingReadOp, but it forgets to remove the cursor from managedledger.waitingCursors
> ```
>
> https://github.com/apache/pulsar/pull/24551/files#diff-1d721da3aaff7689077d33bf3e55fd8d6a5c9184549857a6120d8050c5d67798L385-L386
>
> Here, try to add another one.
Removing from the waitingCursors is a relatively costly operation due to the data structure which is a ConcurrentLinkedQueue in ManagedLedgerImpl. That is most likely why it hasn't been done in the original code. Instead of removing it, it's better to rely on ManagedLedgerImpl.notifyCursors to unqueue the waiting cursor. Therefore, the more optimal solution is to track whether there's a waiting cursor already in the queue when it has been requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2235684079
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1046,7 +1045,7 @@ public void testResumptionAfterBacklogRelaxed() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 15000)
+ @Test(timeOut = 30000)
public void testCloseReplicatorStartProducer() throws Exception {
Review Comment:
The issue you described has been fixed by https://github.com/apache/pulsar/pull/24551, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237201188
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -354,6 +354,13 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
if (dispatcher != null && dispatcher.getConsumers().isEmpty()) {
deactivateCursor();
+ // Remove the cursor from the waiting cursors list.
+ // For durable cursors, we should *not* cancel the pending read. This is because internally,
+ // in the dispatcher implementations, there is a "havePendingRead" flag that is not reset. If the pending
+ // read is cancelled, the dispatcher will not continue reading from the managed ledger when a new
+ // consumer is added to the dispatcher since based on the "havePendingRead" state, it will continue to
+ // expect that a read is pending and will not submit a new read.
+ topic.getManagedLedger().removeWaitingCursor(cursor);
Review Comment:
> The code here does not check if the cursor is durable.
I'll improve the comment. For non-durable cursors, it would be possible to call `cancelPendingReadRequest`. For durable cursors, that shouldn't be called since it would cause the state of the havePendingRead flag in dispatcher to get out of sync.
There's existing tests that fail if `cancelPendingReadRequest` would be called.
The `removeWaitingCursor` calls were removed in #24551 changes. Before that, the removeWaitingCursor was called in different location for the non durable and durable cursors. Since the call should be made for both, it's better to have it before the conditional logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix ManagedCursor state management race conditions and lifecycle issues [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24569:
URL: https://github.com/apache/pulsar/pull/24569#discussion_r2237315751
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java:
##########
@@ -3629,27 +3638,40 @@ void asyncDeleteCursorLedger() {
}
private void asyncDeleteCursorLedger(int retry) {
- STATE_UPDATER.set(this, State.Closed);
+ State previousState = changeStateIfNotDeletingOrDeleted(State.Deleting);
+ if (previousState == State.Deleted) {
+ log.warn("[{}-{}] Cursor ledger is already deleting or deleted. state={}", ledger.getName(), name, state);
Review Comment:
@BewareMyPower Thanks, I've improved this in the most recent changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org