[Bug] Shared subscription stalls forever: havePendingRead=true while the cursor has no read (pendingReadOps=0, waitingReadOp=false) — the #26174 invariant, unguarded on PersistentDispatcherMultipleConsumers

### 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**
- Pulsar **4.0.12** (`pulsar version`), BookKeeper 4.17.3
- OpenJDK 64-Bit Server VM **17.0.15+6-Ubuntu-0ubuntu120.04**, mixed mode, sharing
- JVM flags: `-Xms10g -Xmx10g -XX:MaxDirectMemorySize=10g -XX:+UseZGC -XX:+PerfDisableSharedMem -XX:+AlwaysPreTouch`
- OS: Ubuntu 20.04 LTS, x86_64, broker runs in an LXD container
- Cluster: 5 brokers in this zone; the affected topic's partition stayed on one broker throughout, and that broker process had 36 days of uptime — no restart, no bundle transfer involved
- Relevant `broker.conf`: **`unblockStuckSubscriptionEnabled=true`** (explicitly enabled, not the default), `managedLedgerReadEntryTimeoutSeconds=0`, `managedLedgerAddEntryTimeoutSeconds=0`

**Topic and subscription**
- Persistent, partitioned topic with 3 partitions; exactly one partition affected
- Subscription type **Shared**, durable, a single connected consumer
- No delayed delivery and no replay in flight at the time (`msgDelayed: 0`, `msgInReplay: 0`)

**Client**
- Go client — an internal fork based on **apache/pulsar-client-go v0.12.2**, built with Go 1.22
- OS: Linux x86_64 (container); one process holds ~25k topics and ~75k partition consumers
- The client is a shared gateway: one Pulsar consumer per topic, shared between worker sessions

### Issue Description

A Shared subscription stopped dispatching mid-traffic and never resumed for **26 hours**, until the partition was manually unloaded. Nothing was logged as an error on either side, the consumer stayed connected the whole time, and the cursor stayed valid.

`topics stats` for the stuck partition, captured before any mitigation:

```json
"msgRateOut": 0.0,
"msgBacklog": 600819,
"unackedMessages": 11,
"blockedSubscriptionOnUnackedMsgs": false,
"type": "Shared",
"consumers": [{
"availablePermits": 1676,
"unackedMessages": 11,
"blockedConsumerOnUnackedMsgs": false,
"msgRateOut": 0.0,
"connectedSince": "<16 days before the stall, unchanged throughout>"
}]
```

`topics stats-internal`, same moment:

```json
"numberOfEntries": 613517,
"lastConfirmedEntry": "L54:2680",
"state": "LedgerOpened",
"waitingCursorsCount": 0,
"pendingAddEntriesCount": 0,
"cursors": {
"<sub>": {
"markDeletePosition": "L1:12697",
"readPosition": "L1:12698",
"waitingReadOp": false,
"pendingReadOps": 0,
"subscriptionHavePendingRead": true,
"subscriptionHavePendingReplayRead": false,
"state": "Open",
"active": false,
"individuallyDeletedMessages": "[]"
}
}
```</sub>

The cursor sits in **L1, the first of 54 existing ledgers**, while `lastConfirmedEntry` is in the 53rd — the cursor is valid, points far behind the tail, and ~600k entries are available to read.

**The inconsistent triple is the bug:**

- `subscriptionHavePendingRead: true` — the dispatcher believes a read is in flight;
- `pendingReadOps: 0` — the managed cursor has no read outstanding;
- `waitingReadOp: false` — and it is not waiting for new entries either.

Consequently `readMoreEntries()` can never issue another read: the flag it checks is set, and nothing exists below that would ever clear it. Every subsequent flow permit, ack and publish is a no-op. This is why we consider it a bug rather than backpressure: the consumer had 1676 unused permits and 600k messages were sitting in the backlog, while the broker issued no read at all.

Expected: a subscription with a non-empty backlog and a consumer holding available permits keeps dispatching.

### Error messages

```text
There are none, and that is part of the report — no WARN or ERROR was logged for this topic on the owning broker for the entire day of the stall (verified with a control query on the same file to confirm the pattern matches). The only recurring evidence is the periodic cursor rollover, which logged an identical frozen pair every 16 minutes for 26 hours:

INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [-partition-0] Updated cursor
<sub>with ledger id md-position=L1:12697 rd-position=L1:12698</sub>

The broker's `unblockStuckSubscriptionEnabled` detector never logged `"Dispatcher is stuck and unblocking by issuing reads"` — zero matches across 8 days of broker logs.
```

### Reproducing the issue

We cannot reproduce it on demand. It has occurred **three times in five weeks** on the same cluster, always on a single partition of a partitioned topic, always Shared subscription with a live consumer, always healthy bookies and no broker restart.

Timeline of this occurrence (UTC):

- `13:22:48` last cursor rollover with an advancing position (in the ledger preceding L1, since trimmed)
- `13:26:38` the consumer's worker session drops on this partition, so the gateway `Nack`s the outstanding message — a redelivery on a subscription with a read in flight
- `13:29:26` **last successful dispatch and ack**; nothing is delivered after this
- `13:37:55` the next rollover already shows the frozen pair, and so does every rollover for the next 26 hours
- `16:06` producers stop publishing on their own; backlog freezes at **600,819** messages / 3.9 GB
- next day `15:47` `topics unload` of this partition → the consumer reconnects and delivery resumes immediately at up to **883 msg/s**; the backlog drains in ~11 minutes with no other action

**Possible trigger.** The session drop at `13:26:38` is the only event correlated with the stall — and the earlier occurrence on this cluster had the same predecessor 3.5 minutes before its stall. A `Nack` is a redelivery, i.e. the same collision #26174 describes for the single-active-consumer dispatcher: redelivery re-arming reads while a read completion is in flight. We cannot prove the race from outside, and such session drops happen dozens of times a day without consequence — so this is a hypothesis, not a diagnosis.

### Additional information

**The `unblockStuckSubscriptionEnabled` detector is enabled here and structurally cannot see this state.** In 4.0.12:

```java
if (isAtleastOneConsumerAvailable() && !havePendingReplayRead && !havePendingRead
&& cursor.getNumberOfEntriesInBacklog(false) > 0) {
log.warn("{}-{} Dispatcher is stuck and unblocking by issuing reads", topic.getName(), name);
readMoreEntriesAsync();
return true;
}
```

The guard requires `!havePendingRead`, so it recovers only the case where the dispatcher *knows* it has no read in flight. The failure reported here is the opposite one — and it is precisely the variant that never recovers on its own.

**Where the flag can be left set.** `havePendingRead` is assigned in four places in this class: set `true` in `readMoreEntries()` before `cursor.asyncReadEntriesWithSkipOrWait(...)`, cleared in `readEntriesComplete()` and `readEntriesFailed()` (both under `readType == Normal`), and conditionally cleared in:

```java
protected synchronized void cancelPendingRead() {
if (havePendingRead && cursor.cancelPendingReadRequest()) {
havePendingRead = false;
}
}
```

If the cursor has no request to cancel — exactly our `pendingReadOps: 0, waitingReadOp: false` — this early-returns and the flag stays set forever. Two candidate paths lead to the observed state, and we cannot tell them apart from outside: a read completion that never reached the dispatcher (or reached it after it had moved on), or `cancelPendingRead()` racing a teardown path.

**Broker thread dump taken while stuck** (840 threads): no deadlock, zero threads waiting on monitors, and **not a single `org.apache.bookkeeper.mledger` or `org.apache.pulsar.broker.service` frame** — nothing was executing in a dispatcher or read path. All `bookkeeper-ml-workers` and `bookkeeper-ml-scheduler` threads were parked with empty queues. This rules out "a read was issued to BookKeeper and never returned": there was no read anywhere.

**Client goroutine profile taken at the same moment**: the connection readers were all alive (one per connection, count matching exactly), and the partition consumer's dispatch goroutines were alive and blocked on their channels with a wait duration matching the stall to the minute. The TCP connection was never re-established, and other partitions on the same connection kept working normally. The client was idle-waiting with permits granted; the broker simply never sent anything.

Questions for maintainers:

1. Should the `readOpEpoch` guard from #26174 be extended to `PersistentDispatcherMultipleConsumers`? What we observed is the same invariant violation that PR restores for the single-active-consumer dispatcher, just in the opposite direction — and unrecoverable rather than transient.
2. Should `checkAndUnblockIfStuck` also cover `havePendingRead == true` with no cursor-level read (`pendingReadOps == 0 && !waitingReadOp`)? Today the one variant that never self-heals is the one the guard excludes.
3. Is the `cancelPendingRead()` early return a known leak, and can it be reached without all consumers being removed? Our consumer stayed connected throughout — `connectedSince` never changed.
4. With `managedLedgerReadEntryTimeoutSeconds=0` there is no timeout to break the state. Would a non-zero value help at all here, given that no read exists at the cursor level to time out?

[broker04b-jstack-20260902T154053Z.txt](https://github.com/user-attachments/files/31748783/broker04b-jstack-20260902T154053Z.txt)

[stats-before-unload.json](https://github.com/user-attachments/files/31748904/stats-before-unload.json)
[stats-internal-before-unload.json](https://github.com/user-attachments/files/31748903/stats-internal-before-unload.json)

Closest existing work, none of it covering this state:

- **#26174** (merged 2026-07-23, cherry-picked to 4.0.13 / 4.2.4) — restores exactly this invariant ("an armed read operation implies `havePendingRead == true`") for `PersistentDispatcherSingleActiveConsumer` by introducing a monotonic `readOpEpoch` so that stale read completions are ignored. That PR states that `PersistentDispatcherMultipleConsumers` keeps its own `havePendingRead` and is *"deliberately out of scope"*. This report is the same invariant broken on `PersistentDispatcherMultipleConsumers` with a **Shared** subscription, in the opposite direction: there the cursor held an armed read while the dispatcher believed none was pending; here the dispatcher believes a read is pending while the cursor holds none — and unlike that case, this state never self-heals.
- **#26236** (merged 2026-07-25, in 4.0.13) — a dispatch stall with `havePendingRead=true`, but in `PersistentStickyKeyDispatcherMultipleConsumers` (Key_Shared, PIP-379 look-ahead, delayed messages + replay). Our subscription is plain Shared, with `msgDelayed: 0` and `msgInReplay: 0`, and there the read is genuinely parked at the tail, whereas here no read exists at the cursor at all.
- **#9789** — added `unblockStuckSubscriptionEnabled`, whose detection condition describes this symptom, but whose guard cannot fire on it (see Additional information).
- **#17493** — same symptom class, reported on 2.10.1, but with `"waitingReadOp": true`; here it is `false`.

On branch-4.0 during 2026 the only commits touching `PersistentDispatcherMultipleConsumers.java` are #26055 (dispatch performance) and #26053 (delayed-delivery tracker close path); neither addresses this state.

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!