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

[merlimat]

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

[lhotari]

Thanks for working on this rollover termination race. The queued-add cleanup paths look good, but the in-flight metadata update still has a version-conflict ordering that can fence the ledger or leave the termination unstored.

[lhotari]

[BUG] The BadVersion path bypasses the terminated-state recovery

asyncTerminate() can finish closing the current ledger and submit its metadata write before the rollover update callback runs (ManagedLedgerImpl.java:1561-1582). Both writes then use the same ledgersStat, so one fails with BadVersionException. In this callback, handleBadVersion(e) runs before the added Terminated check, changes the state to Fenced, and the BadVersion branch returns without unlocking metadataMutex (ManagedLedgerImpl.java:1822-1853). If the rollover write wins, termination fails and its terminated position is not stored; if the termination write wins, this callback changes the in-memory state from Terminated to Fenced. The new test holds the BookKeeper close until after the rollover response is released (ManagedLedgerTerminationTest.java:351-373), so it excludes this ordering. Please serialize the termination metadata write with metadataMutex, or reconcile/retry this expected version conflict, and add the opposite-order regression case.

[merlimat]

Good catch, thanks — the terminate's metadata write was the one writer of the ledgers list not going through metadataMutex. Fixed in c1c7715: the terminated position is now stored under metadataMutex, deferring the write while another update is in flight (same tryLock + 100 ms retry as the rollover and the trimming), and the terminate fails instead of writing if the managed ledger was closed or fenced while waiting.

Added terminateWhileLedgersListUpdateIsInFlight, covering both orders: rollover write applied first (the terminate used to fail with BadVersionException and fence the managed ledger) and terminate write applied first (the rollover callback used to flip the state to Fenced and leak the mutex). With the fix the terminate issues no write until the rollover callback has released the mutex. terminateWhileLedgersListUpdateIsDeferred switched to the async terminate for the same reason, since it holds the mutex itself.