[fix][broker] Cancel queued transaction snapshot recovery on topic close

[Demogorgon314]

Motivation

Transaction snapshot recovery tasks can remain queued after a topic is closed.
The queued task retains the snapshot processor, topic, and managed ledger,
causing closed topics to accumulate when topic loading is repeatedly retried.

Modifications

• Cancel queued snapshot recovery tasks when the transaction buffer closes.
• Wait for running recovery to stop before releasing snapshot resources.
• Prevent recovery from succeeding after close has started.
• Stop segmented recovery between I/O operations after close.
• Handle recovery task submission rejection asynchronously.
• Add tests for queued, running, completed, retried, and rejected recovery.

Topic close may now wait for an in-flight recovery I/O operation to finish before releasing snapshot resources.

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

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[lhotari]

Thanks for taking this on — the underlying problem is real, and the shape of the fix (a per-attempt state machine plus a recoveryStoppedFuture barrier) is the right one. The three points from the earlier review look genuinely addressed: the stopped-future barrier does cover synchronous continuations, recoveryIndexUpdateFuture is safely published (written on the recovery thread before stoppedFuture.complete, read only in a dependent of it), and the isClosed() check is now at readSegmentEntries entry ahead of openReadOnlyManagedLedger. The new tests are deterministic — latch-driven, no sleeps, no reflection — and SnapshotSegmentAbortedTxnProcessorCloseTest genuinely pins the fix rather than passing vacuously.

My concern is not the barrier itself but what ended up behind it. Because recoverFromSnapshot() hands back future.copy() and TopicTransactionBufferRecover attaches a non-async thenAccept, the continuation that runs inside future.complete(...) is not a small callback — it is the entire transaction-buffer replay: open a non-durable cursor and read every entry through to LAC. Waiting for that before releasing resources is correct for safety, but it puts the whole replay inside topic close, and the two most common close types gate the managed-ledger close on it. Two consequences follow that I think need addressing before merge (the first two inline comments); a third is a residual gap in the barrier itself.

One unrelated, pre-existing bug I noticed while reading PersistentWorker, mentioned only so it does not stay buried — it is not something I think you should fix here. In the Clear branch, taskQueue.forEach(pair -> pair.getRight().getRight().get().completeExceptionally(...)): pair.getRight().getRight() is the Supplier, so .get() starts every queued WriteSegment/DeleteSegment instead of cancelling it, and then completes the newly started task's future rather than the stored taskExecutedResult (which stays incomplete forever). Compare executeTask(), which correctly uses pair.getRight().getKey() for the result future. The effect is that topic deletion launches the writes it means to cancel and then deletes the segments concurrently. Worth a separate issue/PR.

[lhotari]

[BUG] failing the recovery future during close re-enters topic.close(true) and escalates a graceful ELM transfer into a forceful close, orphaning the transfer future

This completion runs the caller's chain synchronously on the closing thread: .exceptionally in TopicTransactionBufferRecover.run() → recoverExceptionally (TopicTransactionBuffer.java:213-231) → an unconditional topic.close(true). On the base branch close never completed the recovery future, so this re-entrancy is new.

The re-entrant call is not deduplicated when the outer close is a transfer:

• PersistentTopic.close(boolean) (:1753-1754) delegates to close(true, true), i.e. CloseTypes.notWaitDisconnectClients.
• An outer transferring close installs CloseFutures(new CF, null, null) at :1806, so the guard at :1793 (closeFutures.notWaitDisconnectClients != null) does not fire.
• The nested pass therefore runs in full: it fences again, overwrites this.closeFutures (:1807-1808), disconnects every producer (:1826), and — since notWaitDisconnectClients sets disconnectClientsInCurrentCall = completedFuture(null) (:1870) — calls ledger.asyncClose immediately (:1905), reaching disposeTopic in the middle of the transfer.

There is a second-order effect that looks worse than the disconnect itself. The nested call captures the old transfer future as inProgressTransferCloseTask (:1799-1800) and adds it to its own wait set (:1816-1818), but the outer call later wires closeFutures.transferring (:1913) after the field has been replaced — so it completes the new triple's future and the old one is never completed by anyone. The nested call's waitDisconnectClients future (:1920) is chained on that wait set, so it never completes either, and the ELM's second-phase close(true, false) returns it at :1796.

Failure scenario: ExtensibleLoadManager begins transferring a topic whose transaction-buffer recovery is still queued or running. The nested close(true, true) disconnects clients and closes the managed ledger mid-transfer, defeating the graceful hand-over; the orphaned transfer future then leaves the second phase waiting on something nothing will complete.

The trigger is broader than the queued case: a close during RECOVERY_RUNNING reaches the same place from the recovery thread, via future.completeExceptionally(closedException()) at :89/:92.

Could the recovery future be failed off the closing thread, or recoverExceptionally be made to skip topic.close when the topic is already closing?

[lhotari]

[BUG] topic close now waits for the entire transaction-buffer replay, and the wait is unbounded when cursor reads keep failing transiently

stoppedFuture completes in the finally at :94, which is reached only after future.complete(recoveredPosition) at :87 returns — and that call synchronously runs the caller's thenAccept, which is the whole replay: open a non-durable cursor and read every entry from the snapshot position through to LAC, polling with Thread.sleep(1) (TopicTransactionBuffer.java:838-876). Chaining closeResources() here therefore chains it behind the replay, not behind snapshot I/O.

That reaches the managed-ledger close: PersistentTopic.java:1820 puts transactionBuffer.closeAsync() into futures, and :1905 defers ledger.asyncClose until they all complete — for waitDisconnectClients (the ordinary close(true) unload) and transferring, both of which use FutureUtil.waitForAll(futures) (:1866/:1875). There is no timeout on that path. (notWaitDisconnectClients is unaffected: it closes the ledger without waiting, so the reads fail fast.)

Worse, the replay does not necessarily terminate. FillEntryQueueCallback.readEntriesFailed (TopicTransactionBuffer.java:983-991) clears isReadable only for non-recoverable-with-auto-skip, fenced, and cursor-already-closed. Any other error just decrements outstandingReadsRequests, the read position never advances, hasMoreEntries() stays true, and fillQueue() reissues the read forever. In exactly these two close types the managed ledger is only closed after the barrier, so nothing converts the error into a terminal one.

There is also no way to cut it short: recoveryTask.cancel(false) at :142 fires only for a still-queued task, and the replay loop never consults the processor's or the buffer's closed state.

Failure scenario: a topic with a large post-snapshot backlog is unloaded while its transaction buffer is recovering — close blocks for the full replay and the unload can exceed its timeout. With a recurring transient read error (for example repeated BK read failures that are not fenced/non-recoverable), the replay spins indefinitely, the topic never finishes closing, and the shared per-namespace pulsar-transaction-snapshot-recover thread is held throughout.

The description's "may now wait for an in-flight recovery I/O operation" understates this considerably. Would it be reasonable to let the replay observe close — expose the closed state so the drain loop can bail — rather than having close wait for it unconditionally?

[lhotari]

[BUG] the stopped-future barrier misses a continuation attached after the recovery future has already completed

This finally covers the caller's continuation only when the continuation was attached before future.complete(...) at :87 — that is what makes it run inside complete() and therefore ahead of this line.

TopicTransactionBufferRecover.run() attaches its thenAccept at TopicTransactionBuffer.java:832 strictly after recoverFromSnapshot() has returned, and nothing orders that against the recovery thread. If recovery gets all the way through doRecoverFromSnapshot in that gap, both future and stoppedFuture are already complete when the continuation attaches, and the replay then runs on the transaction-executor thread entirely outside the barrier.

That is not as remote as it looks: TableView.internalReadLatest (TableView.java:70-85) serves a warm namespace from a cached reader plus an in-memory map, so doRecoverFromSnapshot can return a non-null snapshot in well under a millisecond — and a bulk namespace reload, the scenario in the motivation, is exactly when the table view is warm. One scheduling preemption of the transaction-executor thread between the call and the attach is enough.

Failure scenario: recovery completes in that gap; a concurrent closeAsync() observes RECOVERY_FINISHED with stoppedFuture already done and runs closeResources() at once, releasing snapshotSegmentsWriter/snapshotIndexWriter while the replay is still calling putAbortedTxnAndPosition → appendTask(WriteSegment, …) through them. This is the same use-after-release the barrier was introduced to prevent, via an interleaving it does not cover — and testCloseWaitsForSynchronousRecoveryCallback cannot catch it, because it attaches its callback before completion.

The window is narrow, so this may well be follow-up work rather than a blocker — but the barrier is presented as a guarantee and currently is not one. Binding the continuation inside recoverFromSnapshot() (so the attach cannot lose the race) would close it.

[lhotari]

[QUALITY] recovery can still report success after close has already won the race

tryMarkRecoveryFinished() flips the state under the monitor (:118-124), but this completion happens after the monitor is released. A closeAsync() that acquires it in between sees RECOVERY_FINISHED, sets CLOSED, and takes the recoveryOwnsFutureCompletion == true branch (:137) — so it deliberately leaves completion to recovery, and the already-taken true branch then completes it successfully. Recovery therefore reports success on a closed processor, which sits against the stated goal "prevent recovery from succeeding after close has started".

Nothing unsafe follows — closeResources() is still gated on stoppedFuture — so I would not block on this. The practical cost is that a topic that has begun closing still performs a full replay, which is what makes it worth mentioning alongside the timing issue above.

Worth noting what not to do: completing the future inside the synchronized block would fix the window but run the entire replay continuation while holding the processor monitor, which your own verifyRecoveryCallbackDidNotHoldProcessorLock assertion is there to prevent. If you want to close it, the decision and the completion need to be made atomic some other way — or it may be simplest to accept it and document that the state transition is the linearization point.

[lhotari]

[QUALITY] a retried recovery replaces the barrier, so close can release resources under the previous attempt

This gates a new attempt on recoveryFuture, which is already done at :87 while the previous attempt's continuations are still running. A second recoverFromSnapshot() in that window installs a fresh recoveryFuture/recoveryStoppedFuture and sets RECOVERY_QUEUED; a following closeAsync() then sees RECOVERY_QUEUED, completes the new stopped future itself (:146-147) and runs closeResources() immediately — while attempt #1's replay is still writing through the writers.

This is latent today, and I checked before raising it: the only production caller is TopicTransactionBuffer.java:832, reached once per processor from recover() at :147, and a failed recovery closes the topic so a reload builds a fresh buffer and processor. So nothing in-tree can hit it.

I mention it because the class now advertises retry — the comment at :62 and testRetryAfterRecoveryFailed — and recoverFromSnapshot() is interface API. If retry is meant to be supported, the barrier should be per-attempt on the stop side too (wait for all outstanding attempts, not just the newest); if it is not, it may be clearer to reject a second attempt while one is still stopping.

[lhotari]

[QUALITY] the close barrier covers the recovery-initiated index update but not the persistent worker's other in-flight tasks

closeResources() awaits recoveryIndexUpdateFuture and then calls PersistentWorker.closeAsync(), which releases both snapshotSegmentsWriter and snapshotIndexWriter immediately (:849-856) without draining taskQueue or waiting for an operation already in Operating. So a WriteSegment/DeleteSegment in flight at close time — including one enqueued by the replay through putAbortedTxnAndPosition — can still have its writer released underneath it.

Related: PersistentWorker.closeAsync() sets the closed flag, but appendTask never consults it (it only checks OperationState.Closed, which only the Clear op sets), so tasks can still be appended and started after close.

This is pre-existing rather than introduced here — on the base branch closeAsync() delegated straight to persistentWorker.closeAsync() — and the earlier review offered draining the worker as the alternative to the narrower fix you took. I am not asking you to widen the scope; I am recording it so the remaining hole is explicit rather than assumed closed by this PR. Snapshots are reconstructible from the topic ledger, so the impact is bounded.