You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "FlorentinDUBOIS (via GitHub)" <gi...@apache.org> on 2026/10/15 12:00:29 UTC
[PR] [fix][broker] Preserve non-durable cursor ownership during cleanup [pulsar]
FlorentinDUBOIS opened a new pull request, #26925:
URL: https://github.com/apache/pulsar/pull/26925
Main Issue: https://github.com/apache/pulsar/issues/26924
### Motivation
Non-durable `removeConsumer` currently removes a subscription from the topic map before a later executor continuation deletes its named cursor. During that gap a replacement subscription can obtain the cached cursor. The old cleanup then resolves and deletes by name, without proving that the topic still owns the original subscription instance.
The source ordering is visible at [`PersistentSubscription.removeConsumer`](https://github.com/apache/pulsar/blob/c1aaf961086380fbdd7bae734e2fed47e54f7e2a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java#L392-L431), [`PersistentTopic.getNonDurableSubscription`](https://github.com/apache/pulsar/blob/c1aaf961086380fbdd7bae734e2fed47e54f7e2a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L1262-L1293), and [`ManagedLedgerImpl` cursor caching/deletion](https://github.com/apache/pulsar/blob/c1aaf961086380fbdd7bae734e2fed47e54f7e2a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L1188-L1207). This is a source-level race analysis, not an executed regression. The suspected lifecycle race has not yet been reproduced at runtime.
### Modifications
- Keep the retiring non-durable subscription reserved until its cursor-deletion callback completes, using the ledger's existing creation lock and exact-instance map removal.
- Add a monitor-guarded terminal retirement state so a waiting consumer or a later reset cannot reopen the old subscription after a reset callback clears its ordinary fence.
- Serialize reset admission, fence CAS and publication of the in-progress reset future under the subscription monitor. A reset already admitted is allowed to finish before asynchronous retirement; no future is waited on under that monitor. Reset retirement preserves its cursor rather than deleting it.
- Reuse `asyncDeleteCursor`, `FutureUtil.supplySafely` and the existing executor. Deletion callbacks only complete a future; map/statistics continuation runs via `thenComposeAsync`, outside the ledger callback's lock.
- Preserve existing name-only removal callers, durable cleanup behavior, and `closeCursor`/`resumeAfterFence` fence operations. No dependencies, wire format or configuration changes.
**Recovery limitation:** an unexpected deletion error keeps the old subscription terminally fenced and reserved; no automatic recovery is added. Missing-cursor/missing-ledger completions are treated as already deleted. This behavior needs review and native failure-path tests.
### Verifying this change
- [x] Make sure that the change passes the fork CI checks linked below.
- [ ] Add a deterministic native regression observed failing on unpatched code and passing with the proposal.
- [ ] Verify reset success/failure races, waiting/new consumer admission, conditional removal and deletion failure recovery.
Locally, only `git diff --check` was run; no local Java build or native test was executed. Existing fork CI passed for commit `8e23ec454fe74ce2b416f6778505d8b90d1a6730`: [Pulsar CI, including the required aggregate check](https://github.com/FlorentinDUBOIS/pulsar/actions/runs/38723557278) and the separate [Pulsar CI Flaky suite](https://github.com/FlorentinDUBOIS/pulsar/actions/runs/38723557252). These results validate the existing checks; no dedicated lifecycle regression has been added or observed RED/GREEN. The suspected lifecycle race has not yet been reproduced at runtime, and recovery after an unexpected cursor-deletion failure still needs review and tests. **Draft / EDITED-UNVERIFIED for the proposed concurrency fix; not ready to merge.**
The documented local commands below were not run locally. A dedicated regression remains necessary in addition to the existing CI:
```sh
./gradlew quickCheck
./gradlew sanityCheck
./gradlew :pulsar-broker:test \
--tests 'org.apache.pulsar.broker.service.persistent.PersistentSubscriptionTest' \
--tests 'org.apache.pulsar.broker.service.SubscriptionSeekTest' \
-PtestRetryCount=0
./gradlew :managed-ledger:test \
--tests 'org.apache.bookkeeper.mledger.impl.NonDurableCursorTest' \
-PtestRetryCount=0
```
Those existing tests alone do not prove the new concurrency invariant. This patch was prepared on master commit `c1aaf961086380fbdd7bae734e2fed47e54f7e2a`; the current target branch and remaining recovery/coverage gaps must still be reviewed before promoting the draft.
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [x] The threading model — changes asynchronous retirement ordering and serializes reset admission; no new executor or blocking future wait.
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "FlorentinDUBOIS (via GitHub)" <gi...@apache.org>.
FlorentinDUBOIS commented on code in PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#discussion_r4323713628
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1577,7 +1594,16 @@ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
}
CompletableFuture<Void> removeSubscription(String subscriptionName) {
- PersistentSubscription sub = subscriptions.remove(subscriptionName);
+ return removeSubscription(subscriptionName, null);
+ }
+
+ CompletableFuture<Void> removeSubscription(String subscriptionName, PersistentSubscription expected) {
+ PersistentSubscription sub;
+ if (expected == null) {
+ sub = subscriptions.remove(subscriptionName);
+ } else {
+ sub = subscriptions.remove(subscriptionName, expected) ? expected : null;
Review Comment:
Added in `357cfef715d692633cda8afa6920eff5c08d4600`: a test places a replacement under the same name, calls `removeSubscription(name, expected)` with the stale instance, and checks that the replacement remains. It fails if removal is changed to `remove(name)` and passes with the expected-instance check.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1595,6 +1621,37 @@ CompletableFuture<Void> removeSubscription(String subscriptionName) {
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> removeNonDurableSubscription(String subscriptionName, PersistentSubscription expected) {
+ return FutureUtil.supplySafely(() -> {
+ CompletableFuture<Void> cursorDeleted = new CompletableFuture<>();
+ // Match non-durable creation's lock; keep the old subscription reserved until deletion completes.
+ synchronized (ledger) {
+ if (subscriptions.get(subscriptionName) != expected) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ledger.asyncDeleteCursor(expected.getCursor().getName(), new DeleteCursorCallback() {
+ @Override
+ public void deleteCursorComplete(Object ctx) {
+ cursorDeleted.complete(null);
+ }
+
+ @Override
+ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerNotFoundException
+ || exception instanceof CursorNotFoundException) {
+ cursorDeleted.complete(null);
+ } else {
+ cursorDeleted.completeExceptionally(exception);
Review Comment:
Thanks. I kept the failure visible when cursor deletion fails unexpectedly. The cleanup future propagates the original failure, and the expected subscription remains mapped while the cursor may still exist. I documented this contract and added tests for both an asynchronous callback failure and a synchronous throw.
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#discussion_r4322972986
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -396,47 +401,45 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
// remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
topic.getManagedLedger().removeWaitingCursor(cursor);
- if (!cursor.isDurable()) {
- // If cursor is not durable, we need to clean up the subscription as well. No need to check for active
- // consumers since we already validated that there are no consumers on this dispatcher.
- this.closeCursor(false).thenRun(() -> {
- synchronized (this) {
- if (dispatcher != null) {
- dispatcher.close().thenRun(() -> {
- log.info()
- .log("Successfully closed dispatcher for reader");
- }).exceptionally(ex -> {
- log.error()
- .exception(ex)
- .log("Failed to close dispatcher for reader");
- return null;
- });
- }
- }
- }).exceptionally(exception -> {
- log.error()
- .exception(exception)
- .log("Failed to close subscription for reader");
- return null;
- });
-
- // when topic closes: it iterates through concurrent-subscription map to close each subscription. so,
- // topic.remove again try to access same map which creates deadlock. so, execute it in different thread.
- topic.getBrokerService().pulsar().getExecutor().execute(() -> {
- topic.removeSubscription(subName).thenRunAsync(() -> {
- // Also need remove the cursor here, otherwise the data deletion will not work well.
- // Because data deletion depends on the mark delete position of all cursors.
- if (!isResetCursor) {
- try {
- topic.getManagedLedger().deleteCursor(cursor.getName());
- } catch (InterruptedException | ManagedLedgerException e) {
- log.warn()
- .exception(e)
- .log("Failed to remove non durable cursor");
+ if (!cursor.isDurable() && !nonDurableCursorCleanupStarted) {
+ // A reset also retires this subscription; consumers must look up its replacement.
+ nonDurableCursorCleanupStarted = true;
+ CompletableFuture<Void> resetFuture = inProgressResetCursorFuture;
+ CompletableFuture<Void> resetComplete = resetFuture == null
+ ? CompletableFuture.completedFuture(null)
+ : resetFuture.handle((ignored, error) -> null);
+ // Finish an admitted reset before retiring its cursor. Never wait while holding this monitor.
+ resetComplete.thenRunAsync(() -> {
+ this.closeCursor(false).thenRun(() -> {
+ synchronized (this) {
+ if (dispatcher != null) {
+ dispatcher.close().thenRun(() -> {
+ log.info()
+ .log("Successfully closed dispatcher for reader");
+ }).exceptionally(ex -> {
+ log.error()
+ .exception(ex)
+ .log("Failed to close dispatcher for reader");
+ return null;
+ });
}
}
- }, topic.getBrokerService().pulsar().getExecutor());
- });
+ }).exceptionally(exception -> {
+ log.error()
+ .exception(exception)
+ .log("Failed to close subscription for reader");
+ return null;
+ });
+
+ // Topic close can be iterating the map; retire the subscription on the existing executor.
+ CompletableFuture<Void> cleanup = isResetCursor
+ ? topic.removeSubscription(subName, this)
+ : topic.removeNonDurableSubscription(subName, this);
+ cleanup.exceptionally(error -> {
+ log.warn().exception(error).log("Failed to remove non durable cursor");
Review Comment:
**[QUALITY] The warning text no longer matches what failed**
Nit. `cleanup` now also covers the reset path (where the cursor is intentionally kept) and the stats collection that runs after the map entry has been removed, so "Failed to remove non durable cursor" can be misleading. Something like "Failed to retire non-durable subscription" would be more accurate.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1577,7 +1594,16 @@ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
}
CompletableFuture<Void> removeSubscription(String subscriptionName) {
- PersistentSubscription sub = subscriptions.remove(subscriptionName);
+ return removeSubscription(subscriptionName, null);
+ }
+
+ CompletableFuture<Void> removeSubscription(String subscriptionName, PersistentSubscription expected) {
+ PersistentSubscription sub;
+ if (expected == null) {
+ sub = subscriptions.remove(subscriptionName);
+ } else {
+ sub = subscriptions.remove(subscriptionName, expected) ? expected : null;
Review Comment:
**[QUALITY] No test pins the exact-instance removal**
Non-blocking. I replaced `subscriptions.remove(subscriptionName, expected)` with `subscriptions.remove(subscriptionName)` locally, and all three new tests still passed. The tests cover the reservation and the durable guard, but none of them puts a different subscription instance under the same name before the old cleanup runs, so this is the one ownership check that would regress silently. A small unit test on `removeSubscription(name, expected)` with a replacement instance in the map (expecting the replacement to stay) would pin it.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1595,6 +1621,37 @@ CompletableFuture<Void> removeSubscription(String subscriptionName) {
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> removeNonDurableSubscription(String subscriptionName, PersistentSubscription expected) {
+ return FutureUtil.supplySafely(() -> {
+ CompletableFuture<Void> cursorDeleted = new CompletableFuture<>();
+ // Match non-durable creation's lock; keep the old subscription reserved until deletion completes.
+ synchronized (ledger) {
+ if (subscriptions.get(subscriptionName) != expected) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ledger.asyncDeleteCursor(expected.getCursor().getName(), new DeleteCursorCallback() {
+ @Override
+ public void deleteCursorComplete(Object ctx) {
+ cursorDeleted.complete(null);
+ }
+
+ @Override
+ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerNotFoundException
+ || exception instanceof CursorNotFoundException) {
+ cursorDeleted.complete(null);
+ } else {
+ cursorDeleted.completeExceptionally(exception);
Review Comment:
**[QUALITY] An unexpected delete failure leaves the subscription name fenced until the topic is reloaded**
Non-blocking, and you already call this out as a limitation. For non-durable cursors `asyncDeleteCursor` only fails with `CursorNotFoundException` (handled), so I do not expect to hit this branch in practice. But if it ever does (or `asyncDeleteCursor` throws and `supplySafely` converts it), the old subscription stays in the map with the terminal flag set, and every later consumer of that name gets `SubscriptionFencedException` until the topic is unloaded. Since a non-durable cursor lives only in memory, it may be safer to still remove the map entry on that error path (after logging), rather than poisoning the name. Fine to leave as is if you prefer to keep the failure visible.
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#discussion_r4286783655
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1578,6 +1589,37 @@ CompletableFuture<Void> removeSubscription(String subscriptionName) {
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> removeNonDurableSubscription(String subscriptionName, PersistentSubscription expected) {
+ return FutureUtil.supplySafely(() -> {
+ CompletableFuture<Void> cursorDeleted = new CompletableFuture<>();
+ // Match non-durable creation's lock; keep the old subscription reserved until deletion completes.
+ synchronized (ledger) {
+ if (subscriptions.get(subscriptionName) != expected) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ledger.asyncDeleteCursor(expected.getCursor().getName(), new DeleteCursorCallback() {
+ @Override
+ public void deleteCursorComplete(Object ctx) {
+ cursorDeleted.complete(null);
+ }
+
+ @Override
+ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerNotFoundException
+ || exception instanceof CursorNotFoundException) {
+ cursorDeleted.complete(null);
+ } else {
+ cursorDeleted.completeExceptionally(exception);
+ }
+ }
+ }, null);
+ }
+ // Deletion callbacks can run under the ledger lock; never collect subscription stats there.
+ return cursorDeleted.thenComposeAsync(ignored -> removeSubscription(subscriptionName, expected),
Review Comment:
**[P2] Make durable cursor creation honor the retirement reservation**
There is a mixed-durability window after cursor deletion and before this continuation removes the subscription. For a simple same-name subscription, a durable request can call `getDurableSubscription` during that window: `asyncOpenCursor` creates and persists a new durable cursor, then `openCursorComplete` sees the retiring non-durable subscription and rejects the request. That rejection does not remove the newly created cursor. When this continuation subsequently removes the old subscription, the durable cursor remains orphaned and can retain backlog. A later non-durable request can also reuse that cached durable cursor and consequently skip non-durable cleanup on its last consumer removal.
Please make durable admission participate in the reservation before opening its cursor, coordinated under the same ledger lock, while keeping statistics collection outside the lock. A deterministic mixed-durability regression should cover this interval; the two new mocked-ledger contract tests do not exercise it.
This is a source-established interleaving, not a locally reproduced failure.
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "FlorentinDUBOIS (via GitHub)" <gi...@apache.org>.
FlorentinDUBOIS commented on PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#issuecomment-6191889057
@lhotari I haven't been able to add and validate a dedicated regression test for this race because Codex and Claude have flagged and blocked those attempts through their security checks. The existing fork CI passes, but that does not replace a deterministic reproducer and a regression test that fails without the fix and passes with 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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "FlorentinDUBOIS (via GitHub)" <gi...@apache.org>.
FlorentinDUBOIS commented on code in PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#discussion_r4323715467
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -396,47 +401,45 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
// remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
topic.getManagedLedger().removeWaitingCursor(cursor);
- if (!cursor.isDurable()) {
- // If cursor is not durable, we need to clean up the subscription as well. No need to check for active
- // consumers since we already validated that there are no consumers on this dispatcher.
- this.closeCursor(false).thenRun(() -> {
- synchronized (this) {
- if (dispatcher != null) {
- dispatcher.close().thenRun(() -> {
- log.info()
- .log("Successfully closed dispatcher for reader");
- }).exceptionally(ex -> {
- log.error()
- .exception(ex)
- .log("Failed to close dispatcher for reader");
- return null;
- });
- }
- }
- }).exceptionally(exception -> {
- log.error()
- .exception(exception)
- .log("Failed to close subscription for reader");
- return null;
- });
-
- // when topic closes: it iterates through concurrent-subscription map to close each subscription. so,
- // topic.remove again try to access same map which creates deadlock. so, execute it in different thread.
- topic.getBrokerService().pulsar().getExecutor().execute(() -> {
- topic.removeSubscription(subName).thenRunAsync(() -> {
- // Also need remove the cursor here, otherwise the data deletion will not work well.
- // Because data deletion depends on the mark delete position of all cursors.
- if (!isResetCursor) {
- try {
- topic.getManagedLedger().deleteCursor(cursor.getName());
- } catch (InterruptedException | ManagedLedgerException e) {
- log.warn()
- .exception(e)
- .log("Failed to remove non durable cursor");
+ if (!cursor.isDurable() && !nonDurableCursorCleanupStarted) {
+ // A reset also retires this subscription; consumers must look up its replacement.
+ nonDurableCursorCleanupStarted = true;
+ CompletableFuture<Void> resetFuture = inProgressResetCursorFuture;
+ CompletableFuture<Void> resetComplete = resetFuture == null
+ ? CompletableFuture.completedFuture(null)
+ : resetFuture.handle((ignored, error) -> null);
+ // Finish an admitted reset before retiring its cursor. Never wait while holding this monitor.
+ resetComplete.thenRunAsync(() -> {
+ this.closeCursor(false).thenRun(() -> {
+ synchronized (this) {
+ if (dispatcher != null) {
+ dispatcher.close().thenRun(() -> {
+ log.info()
+ .log("Successfully closed dispatcher for reader");
+ }).exceptionally(ex -> {
+ log.error()
+ .exception(ex)
+ .log("Failed to close dispatcher for reader");
+ return null;
+ });
}
}
- }, topic.getBrokerService().pulsar().getExecutor());
- });
+ }).exceptionally(exception -> {
+ log.error()
+ .exception(exception)
+ .log("Failed to close subscription for reader");
+ return null;
+ });
+
+ // Topic close can be iterating the map; retire the subscription on the existing executor.
+ CompletableFuture<Void> cleanup = isResetCursor
+ ? topic.removeSubscription(subName, this)
+ : topic.removeNonDurableSubscription(subName, this);
+ cleanup.exceptionally(error -> {
+ log.warn().exception(error).log("Failed to remove non durable cursor");
Review Comment:
Thanks, agreed. I changed the warning to `Failed to retire non-durable subscription`, since cleanup can fail during work beyond cursor deletion.
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #26925:
URL: https://github.com/apache/pulsar/pull/26925
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "FlorentinDUBOIS (via GitHub)" <gi...@apache.org>.
FlorentinDUBOIS commented on code in PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#discussion_r4312581003
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1578,6 +1589,37 @@ CompletableFuture<Void> removeSubscription(String subscriptionName) {
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> removeNonDurableSubscription(String subscriptionName, PersistentSubscription expected) {
+ return FutureUtil.supplySafely(() -> {
+ CompletableFuture<Void> cursorDeleted = new CompletableFuture<>();
+ // Match non-durable creation's lock; keep the old subscription reserved until deletion completes.
+ synchronized (ledger) {
+ if (subscriptions.get(subscriptionName) != expected) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ledger.asyncDeleteCursor(expected.getCursor().getName(), new DeleteCursorCallback() {
+ @Override
+ public void deleteCursorComplete(Object ctx) {
+ cursorDeleted.complete(null);
+ }
+
+ @Override
+ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerNotFoundException
+ || exception instanceof CursorNotFoundException) {
+ cursorDeleted.complete(null);
+ } else {
+ cursorDeleted.completeExceptionally(exception);
+ }
+ }
+ }, null);
+ }
+ // Deletion callbacks can run under the ledger lock; never collect subscription stats there.
+ return cursorDeleted.thenComposeAsync(ignored -> removeSubscription(subscriptionName, expected),
Review Comment:
Thanks for identifying this mixed-durability interleaving. Commit `5df25f59c2fab92d5339a4df76a89fa377f8cb7f` checks the same-name subscription under the managed-ledger lock before calling `asyncOpenCursor`; it completes the rejection after leaving the lock, so a retiring non-durable subscription still blocks a durable cursor open until map removal proceeds.
`testDurableSubscriptionRejectedDuringNonDurableCursorRetirement` uses a real client consumer and managed ledger. It closes the non-durable consumer, pauses after the real cursor deletion but before map removal, then attempts durable admission and verifies rejection with no cursor left behind. The unchanged test blob failed on the orphan-cursor assertion against unguarded production and passed with this guard.
The focused test passed again on 2026-10-19 against the exact code and test committed as `5df25f59c2fab92d5339a4df76a89fa377f8cb7f` (1 test, 0 skipped, 0 failures); `quickCheck` passed on that same code.
--
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] Preserve non-durable cursor ownership during cleanup [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26925:
URL: https://github.com/apache/pulsar/pull/26925#discussion_r4324187791
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -396,47 +401,45 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor
// remove waiting cursor from the managed ledger, this applies to both durable and non-durable cursors.
topic.getManagedLedger().removeWaitingCursor(cursor);
- if (!cursor.isDurable()) {
- // If cursor is not durable, we need to clean up the subscription as well. No need to check for active
- // consumers since we already validated that there are no consumers on this dispatcher.
- this.closeCursor(false).thenRun(() -> {
- synchronized (this) {
- if (dispatcher != null) {
- dispatcher.close().thenRun(() -> {
- log.info()
- .log("Successfully closed dispatcher for reader");
- }).exceptionally(ex -> {
- log.error()
- .exception(ex)
- .log("Failed to close dispatcher for reader");
- return null;
- });
- }
- }
- }).exceptionally(exception -> {
- log.error()
- .exception(exception)
- .log("Failed to close subscription for reader");
- return null;
- });
-
- // when topic closes: it iterates through concurrent-subscription map to close each subscription. so,
- // topic.remove again try to access same map which creates deadlock. so, execute it in different thread.
- topic.getBrokerService().pulsar().getExecutor().execute(() -> {
- topic.removeSubscription(subName).thenRunAsync(() -> {
- // Also need remove the cursor here, otherwise the data deletion will not work well.
- // Because data deletion depends on the mark delete position of all cursors.
- if (!isResetCursor) {
- try {
- topic.getManagedLedger().deleteCursor(cursor.getName());
- } catch (InterruptedException | ManagedLedgerException e) {
- log.warn()
- .exception(e)
- .log("Failed to remove non durable cursor");
+ if (!cursor.isDurable() && !nonDurableCursorCleanupStarted) {
+ // A reset also retires this subscription; consumers must look up its replacement.
+ nonDurableCursorCleanupStarted = true;
+ CompletableFuture<Void> resetFuture = inProgressResetCursorFuture;
+ CompletableFuture<Void> resetComplete = resetFuture == null
+ ? CompletableFuture.completedFuture(null)
+ : resetFuture.handle((ignored, error) -> null);
+ // Finish an admitted reset before retiring its cursor. Never wait while holding this monitor.
+ resetComplete.thenRunAsync(() -> {
+ this.closeCursor(false).thenRun(() -> {
+ synchronized (this) {
+ if (dispatcher != null) {
+ dispatcher.close().thenRun(() -> {
+ log.info()
+ .log("Successfully closed dispatcher for reader");
+ }).exceptionally(ex -> {
+ log.error()
+ .exception(ex)
+ .log("Failed to close dispatcher for reader");
+ return null;
+ });
}
}
- }, topic.getBrokerService().pulsar().getExecutor());
- });
+ }).exceptionally(exception -> {
+ log.error()
+ .exception(exception)
+ .log("Failed to close subscription for reader");
+ return null;
+ });
+
+ // Topic close can be iterating the map; retire the subscription on the existing executor.
+ CompletableFuture<Void> cleanup = isResetCursor
+ ? topic.removeSubscription(subName, this)
+ : topic.removeNonDurableSubscription(subName, this);
+ cleanup.exceptionally(error -> {
+ log.warn().exception(error).log("Failed to remove non durable cursor");
Review Comment:
Thanks, the new wording fits the broader cleanup.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1577,7 +1594,16 @@ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
}
CompletableFuture<Void> removeSubscription(String subscriptionName) {
- PersistentSubscription sub = subscriptions.remove(subscriptionName);
+ return removeSubscription(subscriptionName, null);
+ }
+
+ CompletableFuture<Void> removeSubscription(String subscriptionName, PersistentSubscription expected) {
+ PersistentSubscription sub;
+ if (expected == null) {
+ sub = subscriptions.remove(subscriptionName);
+ } else {
+ sub = subscriptions.remove(subscriptionName, expected) ? expected : null;
Review Comment:
Thanks, this is what I was after. I checked it by swapping the removal to `subscriptions.remove(subscriptionName)` locally and the new test fails there, so the ownership check is now pinned.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1595,6 +1621,37 @@ CompletableFuture<Void> removeSubscription(String subscriptionName) {
return CompletableFuture.completedFuture(null);
}
+ CompletableFuture<Void> removeNonDurableSubscription(String subscriptionName, PersistentSubscription expected) {
+ return FutureUtil.supplySafely(() -> {
+ CompletableFuture<Void> cursorDeleted = new CompletableFuture<>();
+ // Match non-durable creation's lock; keep the old subscription reserved until deletion completes.
+ synchronized (ledger) {
+ if (subscriptions.get(subscriptionName) != expected) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ledger.asyncDeleteCursor(expected.getCursor().getName(), new DeleteCursorCallback() {
+ @Override
+ public void deleteCursorComplete(Object ctx) {
+ cursorDeleted.complete(null);
+ }
+
+ @Override
+ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerNotFoundException
+ || exception instanceof CursorNotFoundException) {
+ cursorDeleted.complete(null);
+ } else {
+ cursorDeleted.completeExceptionally(exception);
Review Comment:
Makes sense, keeping the failure visible is a reasonable choice and the documented contract plus the two failure-path tests pin it down. Fine to leave as is.
--
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