[fix][broker] Prevent stale read completions from stranding Failover subscriptions

[alexandrebrg]

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 readEntriesFailedcheck 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 [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

• 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 [Bug][broker] Failover subscription stops dispatching forever after active-consumer handover #26164

[Denovo1998]

Could we make the final assertion behavioral by reconnecting a consumer and verifying that the newly published entry reaches sendMessages (or that the cursor advances)?

assertFalse(afterWake.stranded()) only rejects one exact internal tuple. It could pass for a different broken state, and at this point there is no active consumer, so the publish is not expected to dispatch anything even in the healthy case.

A reconnect + successful delivery assertion would match the externally observable failure reported in #26164 and make this regression test less coupled to the selected implementation.

[Denovo1998]

Could this exercise subscription.removeConsumer(consumer, false) replace manually replaying the two selected lifecycle steps?

The failure depends on the ordering around dispatcher removal, cursor deactivation, and removeWaitingCursor. Calling dispatcher.removeConsumer followed directly by ledger.removeWaitingCursor bypasses deactivateCursor() and makes the regression test easier to drift from the production lifecycle.

If ConsumerStats mocking is the blocker, a small test seam would still be preferable to duplicating the lifecycle sequence here.