alexandrebrg opened a new issue, #26164: URL: https://github.com/apache/pulsar/issues/26164
### Search before reporting - [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar. ### Read release policy - [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker. ### User environment - **Broker:** Apache Pulsar **4.2.0** (official binary distribution); previously observed on **4.1.1** — both contain #24551 (`48dc9ec621`) and #24569 (`c96f27a4dd`) - **Broker OS/HW:** Linux x86_64 (3-broker production cluster, brokers colocated with Pulsar proxies) - **Broker Java:** OpenJDK 21 - **Metadata store:** ZooKeeper 3.9.x · **BookKeeper:** 4.x, E=3/Qw=3/Qa=2 - **Load manager:** `ExtensibleLoadManagerImpl` + `TransferShedder` (frequent bundle transfers) - **Relevant broker config:** `transactionCoordinatorEnabled=false`, `unblockStuckSubscriptionEnabled=false` (default), `managedLedgerNewEntriesCheckDelayInMillis=10` (default), `activeConsumerFailoverDelayTimeMillis=1000` (default) - **Topology of the affected subscription:** single-partition persistent topic, **Failover** subscription, exactly one consumer, low traffic (topic frequently idle at the tail) - **Client:** Rust client ### Issue Description **What happened.** Periodically (weeks apart, always after a Failover active-consumer), the subscription permanently stops receiving messages. The consumer is connected and credited, the backlog is non-zero and growing, and the broker dispatches nothing until the topic is unloaded or the consumer session is recreated. When restarting the consumer, it consumes the backlog again, and will again be stuck once catch-up the initial backlog. Topic stats properly say that the consumer got possibly backlog: ``` type=Failover consumers=1 (active) availablePermits=2000 msgBacklog=1 msgBacklogNoDelayed=1 unackedMessages=0 msgRateOut=0 msgOutCounter=0 lastConsumedTimestamp=0 ``` Topic internal stats seem to show `waitingReadOp` to be stuck at `true` ``` ledger: lastConfirmedEntry = 74106:10 waitingCursorsCount = 0 cursor: markDeletePosition = 74106:5 readPosition = 74106:6 <- correctly positioned, entries 74106:6..10 exist waitingReadOp = true pendingReadOps = 0 state = Open individuallyDeletedMessages = [] ``` **The contradiction is the bug:** the cursor holds an armed waiting read (`waitingReadOp=true`) but is **not present in `ManagedLedgerImpl.waitingCursors`** (`waitingCursorsCount=0`). Every subsequent publish runs `ManagedLedgerImpl.notifyCursors()`, polls an empty queue, and never wakes this cursor. In our capture the waiting read was armed right after a consumer handover (previous consumer acked up to 74106:5 and disconnected; replacement connected ~46 s later and armed the tail read at 74106:6); **five entries (74106:6..10) were then published over ~35 minutes and every one of them failed to wake the cursor** (`entriesAddedCounter − messagesConsumedCounter = 5`). Registration is durably broken, not a single missed notify. Durable cursor metadata is sane throughout — this is in-memory bookkeeping, not cursor corruption. **What we expected.** #24551 introduced the invariant its own test asserts (`testWaitingCursorsCountAfterSwitchingActiveConsumers`: after active-consumer switching, `waitingCursorsCount == 1`), i.e. an armed tail-wait is always registered for notification. We expected that invariant to hold on 4.1.1/4.2.0. **Why it can never self-recover (verified on the v4.2.0 sources).** Once the state exists, every path out is a dead end: 1. `PersistentDispatcherSingleActiveConsumer.readEntriesFailed()`: a new read attempt that collides with the leftover armed op fails with `ManagedLedgerException.ConcurrentWaitCallbackException`, and the handler **returns without rescheduling**: ```java if (exception instanceof ConcurrentWaitCallbackException) { // At most one pending read request is allowed when there are no more entries, we should not trigger more // read operations in this case and just wait the existing read operation completes. return; } ``` …but the "existing read operation" is orphaned and will never complete. 2. `scheduleReadOnActiveConsumer()` early-returns when `havePendingRead` is still true after `cancelPendingRead()` fails to cancel (i.e. `cursor.cancelPendingReadRequest()` returned false), so a consumer re-add issues **no** read. 3. `checkAndUnblockIfStuck()` (from #9789) requires `!havePendingRead`: ```java if (isConsumerAvailable && !havePendingRead && cursor.getNumberOfEntriesInBacklog(false) > 0) { ... } ``` Here the dispatcher believes a read is pending, so even with `unblockStuckSubscriptionEnabled=true` the watchdog declines to act. 4. `ManagedCursorImpl.cancelPendingReadRequest()` clears `waitingReadOp` but does not touch `waitingCursors` membership / `registeredToWaitingCursors`; conversely `checkForNewEntries()` registers via a path guarded by `registeredToWaitingCursors`. The armed-op state and the queue-membership state are maintained by different code paths with no common lock, and a Failover handover exercises both concurrently (`PersistentSubscription.removeConsumer()` — added by #24569 — removes the cursor from `waitingCursors` on last-consumer disconnect while deliberately leaving the pending read armed). **Why we believe this is a bug (and where).** The three coupled states — `ManagedCursorImpl.waitingReadOp`, `waitingCursors` membership tracked by `registeredToWaitingCursors`, and the dispatcher's `havePendingRead` — can desynchronize across an active-consumer handover. #24569's own description names this exact hazard ("a cursor could have a pending read operation registered without being in the waitingCursors collection, preventing proper notifications"); the fix narrowed the window but a residual interleaving survives it. We have not yet pinned the single minimal interleaving (a heap dump at the next occurrence will read `registeredToWaitingCursors` to discriminate between (a) stale-flag `true` making re-registration a no-op vs (b) leftover armed op making the new arm fail via `ConcurrentWaitCallbackException` after which the dispatcher gives up); we will attach that when captured. Either way, the recovery dead-ends above make any such transient desync **permanent**, which is the part we'd argue is clearly fixable. **Recovery (workaround):** Unloading the topic, and if not working, restarting the broker ### Error messages ```text No error message nor logs found related to this ``` ### Reproducing the issue No minimal standalone reproducer yet; most handovers survive it. Conditions under which we hit it in production, weeks apart: 1. Single-partition persistent topic, **Failover** subscription, one consumer, **low/bursty traffic** so the dispatcher frequently arms a tail-wait (`asyncReadEntriesWithSkipOrWait`). 2. **Frequent active-consumer handovers**: consumer reconnects and/or bundle transfers (`ExtensibleLoadManagerImpl` + `TransferShedder` in our case; ~6 cursor-ledger rewrites/day on the affected subscription). 3. Eventually a handover lands in the race window: the replacement consumer's tail-wait is armed but never registered in `waitingCursors`. From then on the subscription is dead to new publishes (stats above) until unloaded. ### Additional information _No response_ ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
