[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 any entries left in the queue on every abnormal exit. A read
  issued by fillQueue() can still be outstanding when the replay ends, so the queue may hold entries
  whose pooled buffers would otherwise never be released. This also covers the pre-existing
  catch (Exception) exit.
• PendingAckHandleImpl.exceptionHandleFuture(): do not retry once the handle is closing or closed. The
  retry path resets the state to None before scheduling init(), which defeats the checkIfClose()
  guard in initPendingAckStore() and reopens the pending ack store of a subscription that is going
  away. This was reachable before via the pre-loop close check; the new in-loop check makes it reachable
  on the ordinary topic unload path.

Verifying this change

• Make sure that the change passes the CI checks.

Three 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.

MLPendingAckStoreTest, the rest of org.apache.pulsar.broker.transaction.pendingack.*, and
TransactionTest all pass locally.

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.

Follow-ups deliberately left out of this PR

Tracked in #26368:

1. TopicTransactionBuffer has the opposite gap: it has the else escape but no cursor.isClosed()
   check anywhere and a bare //no-op InterruptedException handler. Its abort path closes the topic,
   so it deserves its own change and test.
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.