[fix][txn] End pending ack replay on read failures it does not classify

[lhotari]

Fixes #26374

Important

Based on #26369. That PR's commit appears in this diff until it merges — please review only the
top commit, 52f030b. This change composes with the
stopReplay() machinery #26369 introduces and is not separable from it.

Motivation

MLPendingAckStore.FillEntryQueueCallback.readEntriesFailed recognises three exception shapes and ends
the replay for them. Everything else only logged an ERROR and decremented the outstanding read counter,
so fillQueue() immediately re-issued the identical read.

Two classes fall through, both reachable with the shipped defaults:

• NonRecoverableLedgerException when autoSkipNonRecoverableData is false. A deleted or missing
  pending-ack ledger arrives as LedgerNotExistException, and OpReadEntry only advances past a bad
  ledger when auto-skip is enabled — so the read position never moves and every retry is byte-identical.
  Retrying can never succeed.
• A plain ManagedLedgerException, e.g. BookKeeper's BookieHandleNotAvailableException (code
  -8). Retrying is correct here, but not at this rate.

The result is a hot loop of ~1000 reads/second, one ERROR line each, holding a transaction replay
thread. Those executors are single-threaded and assigned by identity hash, and
PersistentSubscription.addConsumerInternal waits on pendingAckHandleFuture() with no timeout, so
every other subscription on that thread stops being able to add consumers.

The reporter of #26364 measured ~47.6k / 1.5k / 59.4k / 31.1k of these ERRORs across four brokers in 15
minutes during a BookKeeper outage, and reproduced the deleted-ledger case deterministically: an
unrelated topic sharing the replay thread also stopped accepting subscriptions.

This is the unfinished half of #12700. That PR set out to stop exactly this — "if any ledger was
deleted from bookkeeper ... MLPendingAckStore will not stop recovering and continue to report the
exception" — and added both the guard and TransactionTest.testEndTPRecoveringWhenManagerLedgerDisReadable.
But its test enables autoSkipNonRecoverableData, so the default configuration was left spinning.

Modifications

• readEntriesFailed ends the replay for every failure. The three shapes it already recognised keep
  completing the replay, exactly as the existing test pins. Any other failure is recorded and the attempt
  is reported through replayFailed, letting PendingAckHandleImpl.exceptionHandleFuture decide what
  happens next — it already classifies correctly: transient failures reschedule init() with backoff
  (freeing the replay thread between attempts), non-recoverable ones fail the subscription rather than
  retrying a read that cannot succeed.
• fillQueue() checks isReadable after outstandingReadsRequests. readEntriesFailed clears
  isReadable before its decrement, so a thread observing the decrement must also observe the cleared
  flag — no read can be issued after a failure. The reverse operand order would permit one more.
• A failed attempt rewinds the cursor. Reads run ahead of processing, so entries read but never
  applied have already advanced the shared cursor's read position and are released when the attempt
  ends. The cursor is cached by the managed ledger, so without a rewind the next attempt would resume
  past those entries, skip them, and report the incomplete replay as successful. (cursor.rewind() sets
  the read position to the first entry after the mark-delete position — exactly where the next attempt's
  currentLoadPosition starts. Re-reading already-applied entries is safe: the replay handlers
  deduplicate.)
• A failed attempt closes its buffered writer. The store is abandoned afterwards, and with
  transactionPendingAckBatchedWriteEnabled its flush task reschedules itself forever with nothing left
  able to stop it.
• The per-failure log is split. Failures that complete the replay log at WARN — they never reach
  exceptionHandleFuture, and the handle goes on to log recovery as a success; a cursor closed between
  two reads fails synchronously with no log at any layer, so this was otherwise invisible. Failures
  reported downstream drop to DEBUG, where exceptionHandleFuture writes one WARN per paced attempt or
  an ERROR when it gives up.

Note

Operator-visible change. The "MLPendingAckStore of topic ... stat reply fail!" line is gone — it
was the string operators grepped for. The equivalents are the new WARN and the messages from
exceptionHandleFuture. Separately, a permanently unreadable pending-ack log now fails subscribe
fast instead of hanging forever; that matches what the same exception already does at store-open time.
Remedy for a genuinely lost ledger: enable autoSkipNonRecoverableData and reload the topic.

Verifying this change

• Make sure that the change passes the CI checks.

Added to MLPendingAckStoreTest, each verified to fail when its corresponding source change is
reverted:

• testReadFailureEndsReplayAttempt — over LedgerNotExistException, a plain ManagedLedgerException
  and TooManyRequestsException: the original exception reaches exceptionHandleFuture unwrapped
  (isRetryableException discriminates on the concrete class), exactly one read is issued, the replay
  thread is released, the replay is never also reported complete, and the cursor is rewound.
• testReadFailuresThatCompleteTheReplay — over the three recognised shapes: each still completes, none
  reaches exceptionHandleFuture, and the cursor is not rewound. This pins the discriminating shape:
  routing every failure to replayFailed fails it (that undiscriminating form was tried and reverted
  while preparing [fix][txn] Stop the pending ack replay loop from spinning forever #26369).
• testFailedReplayAttemptClosesBufferedWriter — the flush timer is cancelled.

MLPendingAckStoreTest (16) and TransactionTest (36) pass locally, including
testEndTPRecoveringWhenManagerLedgerDisReadable.

Does this pull request potentially affect one of the following parts:

• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The threading model: (yes) — a failed replay attempt now releases its transaction replay thread
  and is retried on a timer with backoff, instead of occupying the thread in a read loop.
• The binary protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• doc-not-needed

Follow-ups deliberately left out

• [Bug] TopicTransactionBuffer recovery hot-loops on unclassified read failures and reports partial recovery as complete #26379 — TopicTransactionBuffer has the identical guard and hole, plus a dead exceptionNumber
  counter. Its only failure path closes the whole topic, so it needs its own design and tests rather
  than a mechanical port of this classification.
• Store lifecycle on the terminal path. When exceptionHandleFuture gives up it replaces
  pendingAckStoreFuture with a failed future without closing the real store, so the pending-ack cursor
  and managed ledger stay open in the factory cache. Pre-existing, and riskier to fix than it looks — a
  close racing the next init() can hand the new attempt a mid-close managed ledger whose reads fail
  Fenced, which this code maps to complete. Issue to follow.

Assisted-by: Claude (Opus 5, Fable), OpenAI Codex (gpt-5.6-sol)