[fix][ml] Fail queued adds when a managed ledger is terminated during a ledger rollover - #26680

Motivation

ManagedLedgerImpl.asyncTerminate() sets the Terminated state and closes the current ledger regardless of a ledger rollover that may be in progress (CreatingLedger, or the update of the ledgers list that follows the creation). The rollover callbacks did not check for the terminated state either: createComplete() only special-cased the Closed state, and then either reset the state to ClosedLedger / WriteFailed when the creation failed, or went on to updateLedgersIdsComplete(), which unconditionally set LedgerOpened and re-submitted every queued OpAddEntry to the new ledger.

So a managed ledger terminated while a new ledger was being created became writable again, and the adds queued during the rollover were persisted after the terminated position stored in the metadata. The reproducer sees entries written at 4:0 after the ledger was terminated at 3:0, and the ledger keeps rolling over from there. Readers stop at the terminated position, so those entries are acknowledged to the producer but never delivered.

Scalable topics seal a segment with PersistentTopic.terminate() under producer load, where a rollover in progress is a common state to hit. This is the companion of #26678, which covers the adds that were already in flight on the ledger closed by the terminate (ledgerClosed()): this PR covers the adds that were queued waiting for the next ledger. The two changes touch different code paths.

Modifications

• ManagedLedgerImpl.createComplete(): if the managed ledger was terminated while the ledger was being created (whether the creation succeeded, failed or timed out), abort the rollover instead of reopening the managed ledger for writes.
• The ledgers-list update that follows a successful creation checks for the terminated state as well: before each attempt (the update is deferred while metadataMutex is held by another operation, which leaves room for the terminate to complete in between) and in both its success and failure callbacks. Without the check before the attempt, the new ledger would be persisted next to the terminated position, and a terminated managed ledger could later be recovered with a last ledger that no longer exists.
• Aborting the rollover (abortRolloverAfterTerminate()) keeps the Terminated state, fails the queued adds with ManagedLedgerTerminatedException (as internalAsyncAddEntry() does for adds arriving after the terminate), and closes and deletes the ledger that was just created so that it is not leaked.
• asyncTerminate() stored the terminated position without taking metadataMutex, so it could race the ledgers-list update of the rollover it had overtaken, both with the same expected version: whichever write lost failed with BadVersionException and fenced the managed ledger, either failing the terminate without storing its position, or flipping the in-memory state from Terminated to Fenced after the terminate had already succeeded. The terminated position is now stored under metadataMutex, deferring the write while another update of the ledgers list is in flight, as all the other updates of the ledgers list do.

Verifying this change

This change added tests and can be verified as follows.

Eight new tests in ManagedLedgerTerminationTest terminate the managed ledger at each point of a rollover in progress, driven by gates on the mock BookKeeper client and on the metadata store rather than by timing:

• while the creation of the new ledger is pending, and the creation then succeeds, fails, or times out;
• while the update of the ledgers list is deferred because metadataMutex is held;
• while the update of the ledgers list is in flight and completes, successfully or not, before the terminate gets to its own metadata update;
• while the update of the ledgers list is in flight and the terminate gets to its own metadata update before that one completes, in either order the two writes would otherwise be applied.

Each of them asserts that the queued adds fail with ManagedLedgerTerminatedException, that the state stays Terminated, that nothing was written past the terminated position (in memory, in the stored ManagedLedgerInfo, and in BookKeeper, where the created ledger is deleted), that a later add is still rejected, and, for the first one, that the terminated state is what gets recovered on reopen. All of them fail without the fix, and each of the added checks is covered by a test that fails when only that check is removed.

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