[fix][ml] Fail in-flight adds when a managed ledger is terminated - #26678

Motivation

Terminating a managed ledger closes the current BookKeeper ledger, and that close errors out the adds that are still in flight (LedgerHandle.doAsyncCloseInternal fails every un-acked pending add with LedgerClosedException). OpAddEntry.handleAddFailure hands those adds to ManagedLedgerImpl.ledgerClosed(), which had no branch for the Terminated state and returned without touching them, so they were never completed nor failed.

In the broker this leaks PersistentTopic.pendingWriteOps. Any publish that reaches the terminated ledger fails with ManagedLedgerTerminatedException, which fences the topic, and the topic is only unfenced once the pending writes drain to zero. With the leaked ops that never happens, so the topic stayed fenced forever and every subscribe was rejected with Topic is temporarily unavailable (topicFencingTimeoutSeconds is 0 by default, so nothing force-closes it either).

Scalable topics seal segments by terminating them, routinely and under producer load, so this was hit easily: a v5 StreamConsumer that must drain a sealed segment before its children are handed out could never subscribe to the fenced segment and stopped making progress. The broker log shows repeated Attempting to subscribe to a fenced topic for the sealed segment://... topic.

Modifications

• ManagedLedgerImpl.ledgerClosed(): when the ledger is closed in the Terminated state, fail the pending adds with ManagedLedgerTerminatedException, as is already done for the Closed and Fenced states. No new ledger will ever be created to retry them.

Verifying this change

This change added tests and can be verified as follows:

• ManagedLedgerBkTest.testTerminateFailsInFlightAdds (real bookies): suspends the bookie so an add stays in flight, terminates the ledger, and asserts the add fails with ManagedLedgerTerminatedException. Before the fix the add callback never fires.
• BrokerBkEnsemblesTest.testConsumerCanSubscribeAfterTerminateWithInFlightPublish: delays the bookie response so a publish is in flight when PersistentTopic.terminate() runs, sends one more publish to the terminated topic, and asserts the pending write count drains and a consumer can subscribe. Before the fix the count stays at 1 and the subscribe fails with Topic is temporarily unavailable.

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