[Bug] Pending-ack replay hot-loops on read failures it does not classify, monopolizing a transaction replay thread

[lhotari]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

All maintained lines: 4.0 LTS, 4.2 and master. The guard involved dates to a962137f530 (#12700,
first released in 2.10.0).

Reported by @kaminski-dev in discussion #26364, with a reproduction (see below). Split out of #26368;
#26369 fixes a different state in the same loop and does not address this one.

Minimal reproduce step

When a pending-ack log read fails with an exception that MLPendingAckStore.FillEntryQueueCallback
does not recognise, the replay loop re-issues the identical read immediately, forever, monopolising
its transaction replay executor thread.

// MLPendingAckStore.FillEntryQueueCallback.readEntriesFailed
if (managedLedger.getConfig().isAutoSkipNonRecoverableData()
        && exception instanceof ManagedLedgerException.NonRecoverableLedgerException
        || exception instanceof ManagedLedgerException.ManagedLedgerFencedException
        || exception instanceof ManagedLedgerException.CursorAlreadyClosedException) {
    isReadable = false;
}
log.error()...;                              // one ERROR per failed read
outstandingReadsRequests.decrementAndGet();  // ...then fillQueue() issues the same read again

Two exception classes fall through every branch:

1. NonRecoverableLedgerException when autoSkipNonRecoverableData=false (the default). A deleted
   or otherwise missing pending-ack ledger produces LedgerNotExistException
   (isBkErrorNotRecoverable covers NoSuchLedgerExists*, so
   createManagedLedgerException maps it there). Retrying is futile — the data is gone.
2. Plain ManagedLedgerException, e.g. BookKeeper's BookieHandleNotAvailableException (code
   -8), which is not in isBkErrorNotRecoverable. Retrying is correct here, but not at this rate.

Because OpReadEntry.internalReadEntriesFailed only advances past the bad ledger when
autoSkipNonRecoverableData is enabled, the read position never moves and the same read repeats.

Reproduction (@kaminski-dev, attached to discussion #26364): with
numTransactionReplayThreadPoolSize=1, autoSkipNonRecoverableData=false,
managedLedgerReadEntryTimeoutSeconds=0, build ~500 pending acks in an open transaction so the
pending-ack log spans several ledgers, delete one closed ledger the replay must read
(bookkeeper shell deleteledger), then unload the topic. The victim subscription's subscribeAsync
never completes, and a new subscription on an unrelated topic hashed to the same replay thread also
times out. Their run confirms it with
BKException$BKNoSuchLedgerExistsOnMetadataServerException: No such ledger exists on Metadata Server
and the log line "new subscription creation on control topic timed out due to blocked replay lane".

Observed in production too: in one 15-minute window they measured roughly 47.6k / 1.5k / 59.4k / 31.1k
"MLPendingAckStore ... stat reply fail!" ERRORs across four brokers, every sample being
Bookie handle is not available error code: -8, during a BookKeeper outage.

What did you expect to see?

A read failure ends the replay attempt and releases the executor thread. A transient failure is
retried with backoff; a permanently unreadable log fails the handle so the operator sees it.

What did you see instead?

The replay thread re-reads at roughly 1000 attempts/second with one ERROR line each, and every other
subscription hashed onto that thread cannot add consumers, because
PersistentSubscription.addConsumerInternal waits on pendingAckHandleFuture() with no timeout.

Anything else?

This is the unfinished half of #12700. That PR's motivation was literally this problem — "if any
ledger was deleted from bookkeeper, or ManagerLedger was fenced, MLPendingAckStore will not stop
recovering and continue to report the exception" — and it added the guard above plus
TransactionTest.testEndTPRecoveringWhenManagerLedgerDisReadable. But that test sets
autoSkipNonRecoverableData(true), so the deleted-ledger case was only ever fixed for the
non-default configuration. #14781 later added CursorAlreadyClosedException for resource release on
a closing handle.

That matters for the shape of the fix. The existing test pins Ready for exactly three cases:
NonRecoverableLedgerException with auto-skip enabled, ManagedLedgerFencedException, and
CursorAlreadyClosedException. It pins nothing about the two classes above. So a discriminating
fix — record the exception only for the currently-unhandled classes, then call replayFailed(ex)
after the loop instead of replayComplete() — leaves the pinned behaviour untouched and needs no
design discussion. An undiscriminating one (route every isReadable = false exit to replayFailed)
flips all three assertions; that was tried while preparing #26369 and reverted.

Routing to replayFailed gets the right outcome for both sub-cases for free, because
PendingAckHandleImpl.isRetryableException already classifies them:

• plain ManagedLedgerException → retryable → exceptionHandleFuture reschedules init() on a timer
  with backoff, releasing the replay thread between attempts — which is what actually cures the
  cross-subscription starvation, not throttling the log;
• NonRecoverableLedgerException / ManagedLedgerFencedException → not retryable → the handle goes
  to Error and subscribe fails fast instead of hanging. Note this is already what the same
  exception does when it happens at store-open time, so it makes read-time behaviour consistent —
  but it is a client-visible change from "hangs" to "fails", worth a release note.

Related: TopicTransactionBuffer has the identical guard and the identical hole, plus an
exceptionNumber counter that is written and never read. It needs its own fix; its only failure path
closes the whole topic, so it is not a usable precedent for the pending-ack fix.

Are you willing to submit a PR?

• I'm willing to submit a PR!