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