[fix][txn] Stop the pending ack replay loop from spinning forever

[lhotari]

Fixes #26368

Reported by @kaminski-dev in #26364.

Motivation

MLPendingAckStore.PendingAckReplay.run() loops while

lastConfirmedEntry.compareTo(currentLoadPosition) > 0 && fillEntryQueueCallback.fillQueue()

sleeping 1ms whenever the entry queue is empty. The two halves of that condition are measured from
different positions:

• lastConfirmedEntry is a final snapshot of the managed ledger's last confirmed entry taken in the
  constructor, compared against currentLoadPosition, which is seeded from the cursor's mark-delete
  position;
• whether anything can still be read is decided by cursor.hasMoreEntries(), which follows the cursor's
  read position.

When those disagree permanently, fillQueue() issues no read (its hasMoreEntries() gate is false) yet
still returns isReadable == true, the queue stays empty, and the loop sleeps forever — with no
outstanding read and nothing logged. One way to reach that state: the cursor persists a mark-delete
position at the last entry of a ledger, that ledger becomes trimmable, and after a restart
ManagedCursorImpl.recoveredCursor leaves the stored position unrepaired (it only substitutes when
position.getEntryId() == -1) while the read position moves to a later ledger.

The replay executors are single threaded and assigned by identity hash, so a stuck replay also stops
every other subscription sharing that thread from adding consumers: with transactions enabled every
persistent subscription builds a PendingAckHandle, and PersistentSubscription.addConsumerInternal
waits on pendingAckHandleFuture() with no timeout.

The loop could not be stopped either. cursor.isClosed() was checked only before the loop, so a read
whose completion never arrives could not be ended by closing the subscription, and the
InterruptedException handler only logged, so the thread survived ExecutorProvider.shutdownNow().

TopicTransactionBuffer has the same recovery loop and received the corresponding fix in #13739
(7dee63ed707, 2022-01-18), whose commit message describes this exact state:

if cursor.hasMoreEntries() == false && entryQueue.size() == 0, return false and stop recovering.

• If the cursor is cleared when transaction is recovering, there will no entries can be read, but currentLoadPosition < lastConfirmedEntry.

It was never ported to MLPendingAckStore. The escape hatch that MLPendingAckStore used to have was
removed a month earlier by #12700 (a962137f530), which moved cursor.isClosed() out of the loop body.

Modifications

• FillEntryQueueCallback.fillQueue(): when the cursor has no more entries and the queue is drained,
  set isReadable = false so the replay finishes. This mirrors the existing TopicTransactionBuffer
  implementation. Entries at or below the mark-delete position have already been applied, so completing
  here is correct.
• PendingAckReplay.run(): re-check cursor.isClosed() while waiting for entries, so that closing the
  subscription ends a replay whose read completion never arrives. A cursor closed between two reads
  already ends the loop through the existing read failure handling, because that read fails
  synchronously in ManagedCursorImpl#asyncReadEntriesWithSkip.
• PendingAckReplay.run(): on InterruptedException, restore the interrupt flag and end the replay
  through replayFailed() rather than continuing. An incomplete replay must not be reported as
  successful, and the task is now cancellable.
• PendingAckReplay.run(): release queued entries on normal and abnormal replay completion. Stop
  accepting entries under the same lock used by read completion, so late callbacks release their own
  entries without racing the queue drain. Entry deallocation runs outside the lock and the queue keeps
  a single consumer.
• PendingAckHandleImpl.exceptionHandleFuture(): atomically preserve Close when resetting the state
  for retry, and preserve Close on the error path too. A separate close check is insufficient because
  closeAsync() changes the state outside the handle monitor. A late replay failure must not reopen
  the pending ack store of a subscription that is going away.

Verifying this change

• Make sure that the change passes the CI checks (pending for the updated head).

Five regression tests are added to MLPendingAckStoreTest. Each was verified to fail when the corresponding
source change is reverted:

• testReplayCompletesWhenCursorHasNoMoreEntries builds the diverged state and asserts both that the
  replay completes and that a task queued behind it on the same executor still runs — the latter is what
  the stall actually broke.
• testReplayStopsWhenCursorIsClosedWhileWaitingForEntries drops a read completion, closes the cursor,
  and asserts the replay fails and releases the thread.
• testReplayStopsWhenInterrupted drops a read completion, calls shutdownNow(), and asserts the
  thread terminates, that replayFailed is invoked, and that the replay is never reported as complete.
• testEntriesDeliveredAfterReplayEndedAreReleased delivers entries after replay has stopped and
  verifies that the callback releases them.
• testReplayFailurePreservesClosedHandle delivers repeated retryable failures to a closed handle and
  verifies that it remains closed and cannot return to initialization.

MLPendingAckStoreTest, the rest of org.apache.pulsar.broker.transaction.pendingack.*, and
TransactionTest all pass locally after merging master and strengthening the close-state transitions.
quickCheck passes too. The final run used -PtestRetryCount=0; the new closed-handle regression
was also verified to fail with the old error-state transition (expected Close but found Error).

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) — the pending ack replay task now terminates instead of holding its
  executor thread indefinitely, and it responds to interruption.
• The binary protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• doc-not-needed

Matching PR in forked repository

PR in forked repository: not applicable — the change is small and covered by the added unit tests.

Compatibility with recent recovery fixes

Merged master through 6086ba81f3d. #26512 preserves cursor properties, #26474 restores batch ACK
indexes, and #26509 preserves ledger properties. None changes pending-ack replay termination or the
missing-ledger repair condition in ManagedCursorImpl.recoveredCursor, which still only repairs
positions with entryId == -1. #26335 fixes transaction-buffer recovery cancellation separately.

Follow-ups deliberately left out of this PR

Tracked in #26368:

1. Transaction-buffer recovery cancellation and late-entry cleanup have since been addressed by
   [fix][broker] Cancel queued transaction snapshot recovery on topic close #26335. That change does not cover MLPendingAckStore; this pending-ack fix is still needed.
2. A terminal read failure such as LedgerNotExistException (with the default
   autoSkipNonRecoverableData=false) leaves isReadable == true and re-issues the same doomed read.
   Widening that guard needs care: every isReadable = false exit currently falls through to
   replayComplete(), and TransactionTest.testEndTPRecoveringWhenManagerLedgerDisReadable explicitly
   asserts that a fenced managed ledger or a closed cursor leaves the handle Ready. Changing that is a
   deliberate behaviour change and belongs in its own PR.
3. ManagedLedgerImpl.asyncOpenCursor returns a cached, already-open cursor verbatim. If a second
   MLPendingAckStore is ever built over a cursor whose read position is already past the new store's
   last-confirmed-entry snapshot (an unclean close followed by a reload on the same broker), the new
   completion branch fires immediately and reports a replay that applied nothing. The trimmed-ledger
   case this PR targets is genuinely unrecoverable so completing is right, but the branch cannot
   distinguish the two; a DEBUG line records all four positions when it fires. Distinguishing them
   properly belongs with item 4.
4. Consider extending ManagedCursorImpl.recoveredCursor to repair any persisted position whose ledger
   no longer exists, not only positions with entryId == -1, which would prevent the diverged state
   from forming at all.

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

[void-ptr974]

Thanks for the fix. One small cleanup case worth handling: if replay exits while a read is still outstanding, a later readEntriesComplete() can enqueue entries after the replay has already returned, leaving those entries unreleased. A stopped flag could make late successful callbacks release entries directly instead of enqueueing them.

[lhotari]

Good catch — the race is real and I've fixed it. Pushed.

I went a bit further than a bare flag, because on closer look the flag alone is not sufficient and the
obvious way to strengthen it is unsafe:

• A plain stopped flag leaves a check-then-act window. The callback reads stopped == false, the
  replay thread then sets it and drains, and the callback's fill lands after the drain. Same leak,
  smaller window.
• Letting the late callback drain the queue instead is worse. entryQueue is a JCTools
  SpscArrayQueue, whose poll() is documented as single-consumer only and uses a plain load of the
  consumer index. A callback draining concurrently with the replay thread's drain can extract the same
  element twice and double-release it, corrupting the refcount of a buffer that may be shared with the
  entry cache. That is strictly worse than leaking.

So the handover is done under a small lock: readEntriesComplete either enqueues or, if the replay has
stopped, releases its own entries, and that decision is atomic with respect to
{stopped = true; drain}. Entries are released outside the lock, because EntryImpl.deallocate can run
a deallocation callback. The queue keeps exactly one consumer — late callbacks release entries they
never enqueued, so the SPSC contract still holds. The lock is uncontended and only on the replay
teardown path.

Two things I added on top of your report:

1. All four exits now go through one stopReplay() helper, so a future exit path can't miss it.
2. That includes the normal replayComplete() exit, which had no drain at all. It is reachable with
   entries still queued — fillQueue() can return false via isReadable while the queue is non-empty —
   so that path was leaking on master too, independently of this PR.

Also covered by a new test, testEntriesDeliveredAfterReplayEndedAreReleased: it ends the replay with a
read still in flight, then completes that read and asserts every delivered entry was released. It fails
without the handover.

Separately, and not addressed here: the read-failure path itself still re-issues the same doomed read
when the exception isn't one of the three the guard recognises. That is now
#26374, with a reproduction from the original reporter.

[void-ptr974]

Thanks for handling the close-during-replay case. There is still a small check-then-act window here:

1. A retryable failure observes that the handle is not closed.
2. closeAsync() changes the state to Close.
3. The retry path resumes, changes the state to None, and schedules init().

This ordering allows the pending-ack store to be initialized again after closing has started.

[lhotari]

Fixed in be1cecc. The retry reset now uses an atomic state update that preserves Close, replacing the separate close check and state = None assignment. If a retry was already queued when the handle closes, its None → Initializing CAS cannot change Close.

The error transition also preserves Close, preventing a late failure from changing it to Error and enabling a subsequent retry. Added testReplayFailurePreservesClosedHandle for repeated retryable failures after close; it fails with the old error transition (expected Close but found Error). This test covers the late-failure sequence; the atomic update addresses the concurrent ordering you described.

All 71 scoped pending-ack and transaction tests pass locally with retries disabled, along with quickCheck.