alexandrebrg opened a new pull request, #26174:
URL: https://github.com/apache/pulsar/pull/26174
### 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* read
the flag refers to — while the cursor state it mirrors *is*
generation-guarded
(`(op, opReadId)` in `ManagedCursorImpl`).
`internalRedeliverUnacknowledgedMessages` clears `havePendingRead`, rewinds,
and re-arms a fresh
tail-wait read **without cancelling a read that is already in flight**
(`cursor.cancelPendingReadRequest()` can only cancel a *waiting* op). When
the disowned read
later completes, `readEntriesComplete` / `readEntriesFailed` run
`havePendingRead = false` as
their first statement — clearing a flag that now describes the *newer* armed
read. Result: the
cursor holds an armed `waitingReadOp` while the dispatcher believes no read
is pending.
The desync is benign while the cursor remains in `waitingCursors`. It
becomes permanent on the
next last-consumer disconnect: `cancelPendingRead()` short-circuits on the
false flag
(`if (havePendingRead && ...)`) so the armed op survives, and
`PersistentSubscription#removeConsumer` then removes the cursor from
`waitingCursors`. From
there the subscription is stuck forever:
- every subsequent arm CAS-fails on the leftover op
(`ConcurrentWaitCallbackException`), which
`readEntriesFailed` returns on without rescheduling;
- `addWaitingCursor` is only reachable after a successful arm, so the cursor
can never re-enter
`waitingCursors`;
- every publish's `notifyCursors()` polls a queue the cursor is not in.
The consumer stays connected with permits, the backlog grows,
`msgOutCounter` stays 0, and only
a topic unload recovers — exactly the production signature reported in #
(`waitingReadOp=true`, `pendingReadOps=0`, `waitingCursorsCount=0`, cursor
`state=Open`). On
Failover, every redeliver form (ack-timeout, negative ack, explicit
redeliver, reconnect epoch
bump) funnels into `internalRedeliverUnacknowledgedMessages`, so any of
read completion can mint the desync.
### Modifications
- Add a monotonic `readGeneration` (plain `long`, guarded by the dispatcher
monitor) to
`PersistentDispatcherSingleActiveConsumer`, minted alongside the only
`havePendingRead = true` write in `readMoreEntries`.
- Capture the generation into the read-completion continuation; `readEnt
`readEntriesFailed` check it first and **ignore stale completions** (e
dispatcher state mutated, DEBUG log), so a superseded read can no long
read's flag. This restores the invariant *armed ⇒ havePendingRead=true
disconnect-path short-circuit safe: whenever an op is armed,
`cancelPendingRead()` now
actually cancels it.
- Add `PersistentDispatcherSingleActiveConsumerStuckReadTest`: a
deterministic reproduction
using the real `ManagedLedgerImpl` + `ManagedCursorImpl` +
`PersistentDispatcherSingleActiveConsumer`, with the dispatcher's orde
by 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 repro method
fails on current master and passes with the fix; a negative control ru
scenario in the benign order and passes on both. Fidelity notes (deliv
telescoped ack, `newEntriesCheckDelayInMillis=0` as a determinism pin
the class Javadoc.
Notes for reviewers:
- The fix is completion-side. One narrow corner intentionally keeps master's
behavior: if the
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` / `...Classic` keep their own
`havePendingRead`;
their redeliver model is structurally different (replay queue, no rewind +
immediate re-arm),
so they are deliberately out of scope here.
- `readEntriesFailed` is `@VisibleForTesting`; its two test call sites were
updated for the
added parameter. No public API is touched.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
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.
- Run:
`./gradlew :pulsar-broker:test --tests
'org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumerStuckReadTest'
-PtestFailFast=false -PtestRetryCount=0`
- Existing coverage exercised and green:
`PersistentDispatcherSingleActiveConsumerTest`,
`PersistentTopicTest`, `FailoverSubscriptionTest` (incl. the
`waitingCursors` invariant tests
from #24551), `MessageRedeliveryTest`.
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
Fixes #26164
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
