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

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.

All six points from my last review are addressed, and the way they are addressed is better than what I asked for. Rather than widening the close barrier you narrowed it to processor-owned work and made the replay itself a separately cancellable task that observes close.

One new problem, and it is the only thing I would hold on. Taking the transaction buffer's monitor in closeAsync() introduces a lock-ordering inversion against a monitor the buffer already acquires in the other direction, so a replay racing a fenced publish can deadlock two shared broker threads. Detail in the inline comment.

Two smaller notes, neither blocking:

• PersistentWorker is unchanged, so the residual from my earlier comment is still open: putAbortedTxnAndPosition from the replay can enqueue a WriteSegment right up to the moment close takes the buffer monitor, and closeResources() then releases the writers without draining the queue. Pre-existing, still not something I am asking you to widen the scope for — just recording that the new design does not close it either.
• recoverFromSnapshot() and closeAsync() are the state machine, but neither is final (AbstractSnapshotAbortedTxnProcessor.java:57 and :143). Marking them final would stop a subclass from half-overriding the protocol; isClosed() already is.

The lock-order fix landed in the shape I suggested and it holds up on inspection: removeTxnAndUpdateMaxReadPosition no longer touches the topic, all three of its callers invalidate lastDispatchablePosition after leaving the buffer monitor, and testTxnCompletionUpdatesTopicOutsideBufferLock pins that with a Thread.holdsLock assertion across all four replay/commit/abort combinations. The final nit is done as well, and moving the abort path's complete(null) out of the monitor is a bonus improvement.

One edge of the same cycle is still open, and it is one I missed last round rather than anything new here: recoverComplete() and noNeedToRecover() complete transactionBufferFuture while holding the buffer monitor, checkIfTBRecoverCompletely() hands that same future to handleGetLastMessageId, and that continuation can take the topic monitor inline. Detail and a suggested fix are on TopicTransactionBuffer.java:159 — it is the discipline closeAsync() already applies to itself.

Nothing else in the delta needs attention; it is exactly the requested edits plus the test.

Both open points are closed, and I checked the code rather than taking the reply for it.

recoverComplete() now decides the state transition under the monitor and completes transactionBufferFuture after leaving it, on the failure branch as well as the success one:

noNeedToRecover() does the same at TopicTransactionBuffer.java:188-199. The other two completions were already outside a monitor — recoverExceptionally at TopicTransactionBuffer.java:241-244 and closeAsync() at TopicTransactionBuffer.java:756 — and nothing outside this class completes that future, so no path is left that completes it while the buffer monitor is held. Putting the rule on the field itself at TopicTransactionBuffer.java:97-98 is the right place for it.

testRecoveryNotifiesLastPositionQueryOutsideBufferLock pins it rather than leaving it to argument, and both data-provider values earn their place: null drives noNeedToRecover(), a position drives recoverComplete().

One non-blocking flag on the second commit, detail inline: moving handleLowWaterMark past completableFuture.complete(null) is not only a lock move — it changes how far a single low-water-mark trigger drains the backlog. It looks right and it does not reopen the cycle, but nothing pins it and the commit path was left as it was.

The PersistentWorker residual on the segment processor is unchanged and still deliberately out of scope, exactly as I said in my last review.

Otherwise this looks good to me.