[Bug] TopicTransactionBuffer recovery hot-loops on unclassified read failures and reports partial recovery as complete

[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

master and all maintained lines (4.0 LTS, 4.2). The guard involved is byte-identical to
MLPendingAckStore's and dates to the same 2021 PR family.

Sibling of #26374, which reports the same defect in MLPendingAckStore. Filed separately because the
two have very different failure machinery — see below.

Minimal reproduce step

TopicTransactionBuffer.TopicTransactionBufferRecover.FillEntryQueueCallback.readEntriesFailed
classifies only three exception shapes:

@Override
public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
    if (recover.topic.getManagedLedger().getConfig().isAutoSkipNonRecoverableData()
            && exception instanceof ManagedLedgerException.NonRecoverableLedgerException
            || exception instanceof ManagedLedgerException.ManagedLedgerFencedException
            || exception instanceof ManagedLedgerException.CursorAlreadyClosedException) {
        isReadable = false;
    } else {
        outstandingReadsRequests.decrementAndGet();
    }
    recover.callBackException(exception);
}

Anything else — a plain ManagedLedgerException such as BookKeeper's
BookieHandleNotAvailableException (code -8), or a NonRecoverableLedgerException when
autoSkipNonRecoverableData is false (the default) — takes the else branch, which returns
outstandingReadsRequests to zero. The recovery loop's fillQueue() then re-issues the identical
read immediately, and callBackException logs an ERROR for every attempt:

private void callBackException(ManagedLedgerException e) {
    topicTransactionBuffer.log.error().exception(e)
            .log("Transaction buffer recover fail when recovering transaction entry");
    this.exceptionNumber.getAndIncrement();
}

With autoSkipNonRecoverableData=false, OpReadEntry.internalReadEntriesFailed does not advance past
the bad ledger, so the read position never moves and every retry is byte-identical.

What did you expect to see?

A read failure ends the recovery attempt and releases the recovery thread; the outcome reflects
whether recovery actually succeeded.

What did you see instead?

Two problems.

1. Unbounded hot retry. The loop re-reads at roughly 1000 attempts/second with an ERROR per
attempt, occupying its pulsar-transaction-snapshot-recover thread. That scheduler is an
OrderedScheduler keyed by namespace, so co-location is deterministic rather than random: every
other topic in the namespace whose transaction-buffer recovery is queued behind it stays un-recovered,
and an un-recovered TopicTransactionBuffer reports getMaxReadPosition() as EARLIEST, which halts
consumer dispatch on those topics.

2. A terminal read failure is reported as successful recovery. callBackException only logs and
bumps a counter — it never stops the loop and never reaches recoverExceptionally. When the loop does
end (via the three recognised shapes, or once the cursor is exhausted), control falls through to
callBack.recoverComplete(), so the transaction buffer is marked recovered with partial state.

The identical MLPendingAckStore defect was measured in production by the reporter of #26364:
~47.6k / 1.5k / 59.4k / 31.1k "stat reply fail!" ERRORs across four brokers in 15 minutes during a
BookKeeper outage, every sample Bookie handle is not available error code: -8. The transaction-buffer
loop has the same shape and the same trigger.

Anything else?

exceptionNumber is dead code, and it looks like the vestige of the missing mechanism. It is
declared as an AtomicLong and incremented in callBackException — and those are its only two
occurrences in the entire repository. Nothing ever reads it. Its presence suggests the original design
intended to act after some number of read failures, and that half was never written.

Why this is filed separately from #26374. The two classes share the guard but not the recovery
machinery, and that changes the right answer:

• PendingAckHandleImpl has per-handle retry (isRetryableException + Backoff-paced init()
  rescheduling), so [Bug] Pending-ack replay hot-loops on read failures it does not classify, monopolizing a transaction replay thread #26374 can route a transient failure to a paced retry that releases the thread
  between attempts, and a permanent one to a terminal handle error.
• TopicTransactionBuffer has no per-component retry at all. Its only failure route,
  recoverExceptionally, completes the transaction-buffer future exceptionally and calls
  topic.close(true) — closing the entire topic and disconnecting every producer and consumer, with
  recovery only re-running on topic reload.

So "route unclassified failures to the failure path" is not a mechanical port here: deciding whether a
transient BookKeeper blip should close a whole topic (versus today's hot loop, which at least keeps the
topic serving) is a topic-lifecycle design question that deserves its own analysis and tests.

Note also that TopicTransactionBuffer is not affected by the other half of #26368/#26369: its
fillQueue() already has the else { if (entryQueue.size() == 0) { isReadable = false; } } escape,
added by #13739 in 2022. It is only this read-failure classification that is missing.

Existing coverage to keep green: TransactionTest.testEndTBRecoveringWhenManagerLedgerDisReadable
pins the three recognised shapes, exactly as its pending-ack sibling does for #26374.

Are you willing to submit a PR?

• I'm willing to submit a PR!