[fix][broker] Prevent stale read completions from stranding Failover subscriptions - #26174
Merged
lhotari merged 3 commits intoJul 23, 2026
Merged
Conversation
lhotari
reviewed
Jul 11, 2026
…subscriptions On a single-active-consumer (Failover/Exclusive) subscription, internalRedeliverUnacknowledgedMessages clears havePendingRead and re-arms a fresh tail-wait read without cancelling a read that is already in flight (an in-flight read cannot be cancelled). When the disowned read later completes, readEntriesComplete/readEntriesFailed unconditionally clear havePendingRead again while the fresh op stays armed on the cursor: the dispatcher now believes no read is pending while the cursor holds an armed waiting read. The desync is benign while the cursor remains in waitingCursors, but on the next last-consumer disconnect cancelPendingRead() short-circuits on the false flag (the armed op survives) while PersistentSubscription#removeConsumer removes the cursor from waitingCursors. From then on the subscription is permanently stuck: every re-arm CAS-fails on the leftover op (ConcurrentWaitCallbackException is returned without rescheduling) and every publish's notifyCursors() misses the cursor. Only a topic unload recovers it. The cursor side is already generation-guarded ((op, opReadId)); the dispatcher's boolean mirror is not — that asymmetry is the defect. Fixes apache#26164 Signed-off-by: Alexandre Burgoni <alexandre.burgoni@clever.cloud>
alexandrebrg
force-pushed
the
fix-failover-stuck-read-26164
branch
from
July 13, 2026 08:44
f12a486 to
8f682f0
Compare
alexandrebrg
marked this pull request as ready for review
July 13, 2026 09:46
Denovo1998
previously requested changes
Jul 21, 2026
lhotari
reviewed
Jul 22, 2026
lhotari
left a comment
Member
There was a problem hiding this comment.
please address the review comments and update the stale PR description
- Assert the externally observable contract of apache#26164: both tests now reconnect a fresh consumer after the last-consumer disconnect and require the next published message to actually reach Consumer#sendMessages and advance the cursor read position. The internal state tuple is kept only as a secondary diagnostic, so the regression test is no longer coupled to one fix strategy. - Drive the consumer lifecycle through the real PersistentSubscription#addConsumer and removeConsumer(Consumer, boolean) instead of replaying two selected steps by hand, so dispatcher removal, deactivateCursor() and removeWaitingCursor() run in the production order. The spied dispatcher is injected through the existing PersistentSubscription#reuseOrCreateDispatcher test seam; no production code was added for the test. - Replace the hard-coded line-number references in the test Javadoc with durable method-name anchors, which cannot silently rot as the referenced files change. - Record both the stale and the current readOpEpoch on the two stale-read debug log paths, so a suspected recurrence in production shows which epochs disagreed.
lhotari
approved these changes
Jul 23, 2026
Denovo1998
approved these changes
Jul 23, 2026
sandeep-ctds
pushed a commit
to datastax/pulsar
that referenced
this pull request
Jul 31, 2026
…subscriptions (apache#26174) Signed-off-by: Alexandre Burgoni <alexandre.burgoni@clever.cloud> (cherry picked from commit 0679b0d) (cherry picked from commit 8cbbb6e)
sandeep-ctds
pushed a commit
to datastax/pulsar
that referenced
this pull request
Jul 31, 2026
…subscriptions (apache#26174) Signed-off-by: Alexandre Burgoni <alexandre.burgoni@clever.cloud> (cherry picked from commit 0679b0d) (cherry picked from commit 8cbbb6e)
sandeep-ctds
pushed a commit
to datastax/pulsar
that referenced
this pull request
Jul 31, 2026
…subscriptions (apache#26174) Signed-off-by: Alexandre Burgoni <alexandre.burgoni@clever.cloud> (cherry picked from commit 0679b0d) (cherry picked from commit 8cbbb6e)
nodece
pushed a commit
to ascentstream/pulsar
that referenced
this pull request
Aug 28, 2026
…subscriptions (apache#26174) Signed-off-by: Alexandre Burgoni <alexandre.burgoni@clever.cloud> (cherry picked from commit 0679b0d) (cherry picked from commit 8cbbb6e)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
On a single-active-consumer (Failover / Exclusive) subscription, the dispatcher tracks its one
outstanding read with a plain
volatile boolean havePendingRead, with no notion of which readthe flag refers to — while the cursor state it mirrors is generation-guarded
(
(op, opReadId)inManagedCursorImpl).internalRedeliverUnacknowledgedMessagesclearshavePendingRead, rewinds, and re-arms a freshtail-wait read without cancelling a read that is already in flight
(
cursor.cancelPendingReadRequest()can only cancel a waiting op). When the disowned readlater completes,
readEntriesComplete/readEntriesFailedrunhavePendingRead = falseastheir first statement — clearing a flag that now describes the newer armed read. Result: the
cursor holds an armed
waitingReadOpwhile the dispatcher believes no read is pending.The desync is benign while the cursor remains in
waitingCursors. It becomes permanent on thenext last-consumer disconnect:
cancelPendingRead()short-circuits on the false flag(
if (havePendingRead && ...)) so the armed op survives, andPersistentSubscription#removeConsumerthen removes the cursor fromwaitingCursors. Fromthere the subscription is stuck forever:
ConcurrentWaitCallbackException), whichreadEntriesFailedreturns on without rescheduling;addWaitingCursoris only reachable after a successful arm, so the cursor can never re-enterwaitingCursors;notifyCursors()polls a queue the cursor is not in.The consumer stays connected with permits, the backlog grows,
msgOutCounterstays 0, and onlya topic unload recovers — exactly the production signature reported in #26164
(
waitingReadOp=true,pendingReadOps=0,waitingCursorsCount=0, cursorstate=Open). OnFailover, every redeliver form (ack-timeout, negative ack, explicit redeliver, reconnect epoch
bump) funnels into
internalRedeliverUnacknowledgedMessages, so any of them racing an in-flightread completion can mint the desync.
Modifications
readOpEpoch(plainlong, guarded by the dispatcher monitor) toPersistentDispatcherSingleActiveConsumer, minted alongside the onlyhavePendingRead = truewrite inreadMoreEntries.readEntriesCompleteandreadEntriesFailedcheck it first and ignore stale completions (entries released, nodispatcher state mutated, DEBUG log carrying both the stale and the current epoch), so a
superseded read can no longer clear the newer read's flag. This restores the invariant
armed ⇒
havePendingRead == true, which in turn makes the disconnect-path short-circuitsafe: whenever an op is armed,
cancelPendingRead()now actually cancels it.PersistentDispatcherSingleActiveConsumerStuckReadTest: a deterministic reproductionusing the real
ManagedLedgerImpl+ManagedCursorImpl+PersistentSubscription+PersistentDispatcherSingleActiveConsumer, with the dispatcher's ordered executor replacedby a manual task board so the test chooses between the two production-possible arrival
orders of the racing tasks (the stale completion is posted by the BK completion chain, the
redeliver by the client-command thread — both orders occur in production). The consumer
lifecycle runs through the real
PersistentSubscription#addConsumer/removeConsumer(Consumer, boolean), so dispatcher removal,deactivateCursor()andremoveWaitingCursor()happen in the production order; the spied dispatcher is injectedthrough the existing production seam
PersistentSubscription#reuseOrCreateDispatcher.Both tests assert the externally observable contract of [Bug][broker] Failover subscription stops dispatching forever after active-consumer handover #26164: after the last consumer
disconnects and a fresh consumer reconnects, a message published afterwards must actually
reach that consumer's
sendMessagesand advance the cursor — the internal-state tuple iskept only as a secondary diagnostic. The repro method fails on current master and passes
with the fix; a negative control runs the same scenario in the benign order and passes on
both. Fidelity notes (the one remaining
filterEntriesForConsumerstub, needed because thetest publishes raw bytes rather than serialized Pulsar messages; the telescoped ack;
newEntriesCheckDelayInMillis=0as a determinism pin) are documented in the class Javadoc.Notes for reviewers:
redeliver's re-arm bails (e.g. no permits), the disowned read's completion still counts as
current and dispatches — a benign at-least-once duplicate, unchanged from today.
PersistentDispatcherMultipleConsumers/...Classickeep their ownhavePendingRead;their redeliver model is structurally different (replay queue, no rewind + immediate re-arm),
so they are deliberately out of scope here.
readEntriesFailedis@VisibleForTesting; its two test call sites were updated for theadded parameter. No public API is touched.
Verifying this change
This change added tests and can be verified as follows:
PersistentDispatcherSingleActiveConsumerStuckReadTest#testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion— deterministic reproduction of the strand: fails on master, passes with this fix (verified
over repeated forced re-executions, plus the inverse: re-fails when the fix is reverted).
PersistentDispatcherSingleActiveConsumerStuckReadTest#testRedeliverAfterReadCompletionDoesNotStrandCursor— negative control (same staging, benign task order): passes with and without the fix,
showing the arrival order alone is the trigger.
./gradlew :pulsar-broker:test --tests 'org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumerStuckReadTest' -PtestFailFast=false -PtestRetryCount=0PersistentDispatcherSingleActiveConsumerTest,PersistentTopicTest,FailoverSubscriptionTest(incl. thewaitingCursorsinvariant testsfrom [fix][broker] Fix Broker OOM due to too many waiting cursors and reuse a recycled OpReadEntry incorrectly #24551),
MessageRedeliveryTest.Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Fixes #26164