[fix][ml] Prevent writes to empty ledgers closed in metadata

[sakshichitnis27]

Fixes #26074

Motivation

BookKeeper auto-recovery can close an empty head ledger in durable metadata while the broker still holds the same ledger as open in memory. Once the maximum rollover time is reached, the empty-ledger guard prevents the background rollover from healing this state. The first subsequent write can then be acknowledged on a ledger whose durable metadata records lastEntryId=-1, resulting in a stuck backlog and silent message loss.

Modifications

• Check durable BookKeeper metadata when an empty ledger reaches the maximum rollover time.
• Roll to a new ledger when the durable metadata is already closed, while keeping healthy empty ledgers open.
• Queue writes that arrive during the metadata check so they cannot enter the stale write handle.
• Add a real BookKeeper regression test that recreates the durable-closed/in-memory-open mismatch and verifies the first message is written to and read from the new ledger.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• ./gradlew :managed-ledger:test -PtestRetryCount=0
• ./gradlew quickCheck
• ./gradlew sanityCheck

The managed-ledger suite completed 722 tests with no failures. Repository-wide quickCheck and sanityCheck also completed successfully.

Does this pull request potentially affect one of the following parts:

• 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

The change coordinates pending writes with an asynchronous BookKeeper metadata check using the existing managed-ledger executor; it does not add threads or executors.

[lhotari]

Thanks for tackling #26074 — the diagnosis is right and the regression test is a real one. I ran a fairly deep multi-model review of 3e02157; findings below, most important first.

Verified up front, so the rest is read in context:

• ManagedLedgerBkTest.rollEmptyLedgerClosedInMetadata fails on master with AssertionError: [the first entry should be written to a new ledger] Expecting actual: 0L not to be equal to: 0L, and passes 5/5 on this branch. It is a genuine regression test.
• The BookKeeper interaction is correct: withRecovery(false) yields a ReadOnlyLedgerHandle whose asyncClose only unregisters the metadata listener (BK 4.18.0), so the check neither fences nor closes the ledger the broker is writing to, and the handle is closed on every branch.
• The new flag is accessed only under the ML monitor at all five sites; resendPendingAddEntries on an unchanged ledger is sound (currentLedgerEntries == 0 ⇒ every queued op has ledger == null ⇒ no double-send, no double-count, FIFO preserved).

---

1. Writes queued during the check are stranded forever if the topic is terminated mid-check

ManagedLedgerImpl.java:2080

asyncTerminate() sets state = State.Terminated at :1546 and never calls clearPendingAddEntries. Terminated does not override isFenced() (State enum, :313). Every other exit from LedgerOpened during the check window is covered — asyncClose (:1675), setFenced (:4597), setFencedForDeletion (:4606) all clear pending, and a concurrent roll drains via updateLedgersIdsComplete → resendPendingAddEntries. Terminated is the unique gap, and the new guard

if (currentLedger != ledger || STATE_UPDATER.get(this) != State.LedgerOpened) {
    return;
}

returns without draining or failing the queue.

Scenario: first produce to an empty head ledger past max rollover → op appended at :881 and never initiated (so currentLedgerEntries stays 0) → concurrent asyncTerminate() → callback early-returns. The AddEntryCallback never fires and the retained op.data ByteBuf is never released. Nothing recovers it: later adds fail fast at :871 without touching the queue, and checkAddTimeout can't help because managedLedgerAddEntryTimeoutSeconds defaults to 0.

Suggested fix: fail pending adds in that guard when the state is Terminated, or clear them in asyncTerminate.

2. The guard only closes the loss window after managedLedgerMaxLedgerRolloverTimeMinutes (default 240 min)

ManagedLedgerImpl.java:883 and :2063

Both gates require maximumRolloverTimeReached() (:4531); otherwise internalAsyncAddEntry falls through to :915-931 and writes through the stale handle unchanged.

If auto-recovery fences and closes the empty head ledger 10 minutes after it was created and the first produce lands a minute later, nothing checks metadata, the replacement bookies accept the write, and #26074 reproduces exactly as filed — acked producer, backlog stuck at 1, silently lost on the next recovery. The PR body does scope itself to "once the maximum rollover time is reached", but the title and the linked issue promise more. Worth stating the residual window explicitly in the description so it isn't mistaken for a complete fix.

Non-empty ledgers closed by auto-recovery at a lower LAC are outside this PR's scope entirely, which is fine — but also worth naming.

3. Even inside the guarded window, the check is a TOCTOU

ManagedLedgerImpl.java:2091

The callback reads an OPEN metadata snapshot and then resendPendingAddEntries() (:2099) dispatches through the same pre-existing write handle via OpAddEntry.initiate() (:1903-1919). Nothing ties the snapshot to the write — the monitor orders local state, not BookKeeper auto-recovery. Auto-recovery can close the ledger with lastEntryId=-1 in between, and the write is then accepted by replacement bookies and lost.

One observation that we think makes this worth reconsidering rather than just documenting: in the healthy drain, resendPendingAddEntries bumps entries to 1 and currentLedgerIsFull() is already true (the time-based leg, :4512-4522), so the first drained op gets closeWhenDone and rolls the ledger anyway (:1909-1916). So whenever a write is pending, "keep healthy empty ledgers open" lasts exactly one entry — and that one entry still goes out on the suspect handle.

That suggests a simpler and strictly stronger alternative: on the first add to an empty ledger past max rollover time, unconditionally roll to a new ledger instead of reading metadata. Same amortized cost (one ledger creation instead of one metadata read plus a probable creation), no TOCTOU, no new failure mode when the read fails, and no new blocking gate on the add path — it would dissolve findings 1, 3 and 5. The metadata read only buys something in the no-write timer case.

The true root cause is apache/bookkeeper#4812 (fresh bookies at the same ip:port have no fence state and accept the write), so any broker-side fix is a mitigation. Saying so in the description would set the right expectation.

4. The new internalAsyncAddEntry gate — the PR's headline behaviour — is never exercised by the test

ManagedLedgerBkTest.java:800-806

Verified empirically over 5/5 runs on this branch: Rolling over an empty ledger that is closed in metadata (ManagedLedgerImpl@2095) is always logged from the explicit rollCurrentLedgerIfFull() on :803, on a BK worker thread, before addEntry runs. By then the state is ClosedLedger/CreatingLedger, so internalAsyncAddEntry takes the pre-existing queue branch at :889 and never the new gate at :883-887.

So the comment "Trigger the scheduled check and immediately produce to exercise the race with the metadata lookup" describes a race the test does not reach — and the add-path gate is precisely the production entry point for #26074 (idle topic, first write after auto-recovery closed the ledger, no scheduled roll pending).

Dropping the second rollCurrentLedgerIfFull() and letting addEntry alone drive the check would cover it. Worth adding a case with several concurrent adds during the lookup, plus coverage of the failure branch (:2084) and the metadata-still-open drain (:2099) with a non-empty queue.

5. A transient failure of the metadata read fails every queued write on an otherwise healthy ledger

ManagedLedgerImpl.java:2084

clearPendingAddEntries(createManagedLedgerException(exception)) is the only such call site in the class that leaves the ML in State.LedgerOpened — all eight others accompany a transition to Closed/Fenced/ClosedLedger/WriteFailed.

On a ZK read timeout or ClientClosedException, a write that previously just succeeded is now failed back to the producer, and since the flag resets and currentLedgerEntries stays 0, every retry re-issues the check with no backoff. factory.isMetadataServiceAvailable() only guards a known-disconnected service, not a failing read.

Fail-closed is defensible for a data-loss fix, and our reviewers split on whether this is a bug or a reasonable trade-off. But since currentLedgerEntries == 0 is a precondition, the ledger is provably empty — rolling it on an inconclusive check loses nothing and keeps the topic writable, which seems better than failing producers. Either way it should be a deliberate, commented choice.

6. Minor

• Match asyncCreateLedger's defensiveness. checkEmptyLedgerMetadata (:2071-2112) has neither a try/catch around the builder chain nor a timeout arm, while its sibling asyncCreateLedger in the same file has both — try { ...execute()... } catch (Throwable cause) at :4769/:4786 and scheduledExecutor.schedule(... TimeoutException ...) at :4793. The flag is set at :885/:2066 before the call and cleared in exactly one place (:2078). We could not construct a reachable stall (BK's mainWorkerPool is unbounded with rejectExecution == false, and every terminal path in LedgerOpenOp.initiateWithoutRecovery completes the future), so this is consistency hardening rather than a defect — but it's cheap.
• Duplication. The trigger guard is repeated verbatim at :883-887 and :2063-2067 — worth extracting a maybeCheckEmptyLedgerMetadata(). maximumRolloverTimeReached() (:4531) also duplicates currentLedgerIsFull()'s time logic (drift risk). The new field at :288 has no comment stating it is monitor-guarded, unlike the neighbouring ledgerRecheckInProgress.
• Test hygiene. No @Test(timeOut = ...), unlike peers in the class (:570, :601). Given finding 1, a hang is not hypothetical — addEntry waits on an unbounded CountDownLatch (:791), so a regression would hang CI rather than fail crisply.
• First-write latency. Every reactivation of an idle topic (empty ledger past max rollover — the steady state for low-traffic topics) now pays a BK metadata-open round trip before the first entry dispatches, even when healthy. Multiplied across a broker's idle topics on a post-restart reconnect wave.

---

On rollCurrentLedgerIfFull() becoming synchronized — checked, no deadlock

We spent real effort here since it's the riskiest-looking part of the diff, and want to record the negative result so it doesn't get re-litigated.

The synchronized adds no new edge to the lock graph. Both edges it creates already exist on master inside other synchronized ManagedLedgerImpl methods:

New edge	Already present on master via
ML monitor → LedgerHandle.asyncClose(...) (:2045)	asyncClose (:1658 → :1689), asyncTerminate (:1534 → :1552)
ML monitor → BK open-ledger op (:2072)	addEntryFailedDueToConcurrentlyModified (:1923 → :1929), internalAsyncAddEntry → asyncCreateLedger (:910)

Underneath, the calls are safe by construction: LedgerHandle.asyncClose takes no BK lock on the calling thread (doAsyncCloseInternal's first statement is executeOrdered(...), so the synchronized (LedgerHandle.this) block runs on the BK ordered executor), and BK deliberately drops that monitor before invoking callbacks — errorOutPendingAdds sits outside the block, with the comment "the callbacks shouldn't be running under any bk locks". factory.isMetadataServiceAvailable() is a plain field read, and every task dispatched from under the monitor is fire-and-forget, so there is no pool self-deadlock either.

The one three-party candidate we found — BookKeeper.closeLock (a non-fair RRWL, so a queued writer blocks new readers) — dies because CreateBuilderImpl.execute() releases closeLock.readLock() in a finally before returning the future, and Pulsar attaches .whenComplete(...) only afterwards; no thread ever holds closeLock.read while blocking on the ML monitor.

What is real is contention, not liveness: the monitor is now held across the synchronous portion of newOpenLedgerOp().execute() and asyncClose(), serializing synchronized internalAsyncAddEntry behind it and putting shared bookkeeper-ml-scheduler threads into monitor waits. A narrower critical section around just the new branch would avoid widening that window — the entries > 0 path did not previously need the monitor.

---

AI-assisted review. The analysis above was produced with Claude (Opus 5) and OpenAI Codex (gpt-5.6-sol) reviewing independently and then cross-validating each other's findings; every claim was re-checked against the code, and the test runs cited were executed locally. I've reviewed the output and I'm accountable for what's posted here — please push back on anything that looks wrong, since multi-model review is good at breadth but still gets things confidently wrong. Two candidate findings were dropped during cross-validation for exactly that reason.

A note on ASF policy, since it's easy to miss and applies to contributions as much as reviews — see AGENTS.md and the ASF Generative Tooling guidance:

• A human is in the loop and is accountable. Every PR must be submitted by a human contributor who has reviewed and verified the change and takes responsibility for it — the AI assists, the human is accountable.
• No code of incompatible or unknown provenance. Nothing copied verbatim from GPL/AGPL/LGPL, proprietary, or unlicensed sources (including Stack Overflow / blog snippets of unclear licensing); reimplement from specifications or Apache-compatible sources, per the ASF 3rd Party Licensing Policy.
• Consider attributing AI assistance with an Assisted-by: <tool/version> commit trailer if AI tooling helped on this change (Generated-by: is for minimally-modified generated output).

[Denovo1998]

Ultimately, it may still be necessary to introduce a fencing/incarnation mechanism in BookKeeper #4812 that remains valid across the lifespan of bookie processes. This would enable a replaced bookie to identify stale writers, rather than solely relying on fence states stored in the memory of the old process.

[Denovo1998]

This keeps the original silent-loss window open until the maximum rollover time is reached. Auto-recovery can close an empty ledger soon after it is created, but a first publish before the rollover deadline bypasses this check and is sent directly through the stale write handle. Replacement bookies can acknowledge that write while durable metadata remains CLOSED/-1. Should we verify every first write to an empty ledger, or prove why the pre-rollover scenario is safe? Please add a regression test that closes the metadata without advancing the clock.

[Denovo1998]

maximumRolloverTimeReached() returns false when the broker's metadata service is unavailable. At both call sites, a false value causes the metadata check to be skipped, allowing the write to proceed with the current handle. Since auto-recovery may have already closed the ledger via a different metadata session while this broker is disconnected, this turns a safety check into a fail-open path. Could we differentiate between "the deadline has not been reached" and "the state cannot be verified," and either queue or fail the write in the latter case?

[Denovo1998]

LedgerMetadata.isClosed() returns false for IN_RECOVERY, so this branch resends pending writes even while recovery and fencing may already be underway. However, BookKeeper defines IN_RECOVERY as a state where a reader might be recovering the ledger, and any ensemble changes will force the writer to close it. Given this safety requirement, should only an exact State.OPEN allow reuse of the current handle? IN_RECOVERY should be treated as unsafe, and this warrants a regression test using withInRecoveryState().

[Denovo1998]

An OPEN result here is only a cache-backed snapshot and is not atomic with resendPendingAddEntries(). The metadata can become IN_RECOVERY or CLOSED after the read but before the queued writes are sent, and PulsarLedgerManager.readLedgerMetadata() itself reads through MetadataCache.getWithStats(). In that race, the same stale write handle is reused and the original silent-loss condition remains possible. For an expired empty ledger that already has pending writes, could we retire/roll the old handle instead of reusing it after an OPEN snapshot? Otherwise this should be described as a best-effort mitigation rather than prevention.

[sakshichitnis27]

Thanks @lhotari and @Denovo1998 for the detailed review. I addressed the requested changes in commit be9bad3.

Summary:

• The first write to an empty recorded head ledger now verifies durable BookKeeper metadata even before the maximum rollover deadline, while pending writes remain queued.
• Only an exact LedgerMetadata.State.OPEN permits reuse. CLOSED and IN_RECOVERY are treated as unsafe and rolled over.
• Once the configured maximum rollover time is reached, the empty ledger is retired directly instead of reusing it after an OPEN metadata snapshot.
• Metadata-check failures now trigger a safe rollover rather than failing open or repeatedly failing producers.
• Termination during a pending metadata check or rollover fails and releases queued writes.
• The regression coverage exercises the actual first-write path without manually triggering rollover, including OPEN, CLOSED, IN_RECOVERY, concurrent writes, metadata-read failure, configured-deadline rollover, readability, and termination.

Local validation completed successfully:

• Focused regression and compatibility tests
• Complete :managed-ledger:test suite with retries disabled
• ./gradlew quickCheck
• ./gradlew sanityCheck

This remains a Pulsar-side mitigation for #26074. The underlying replaced-bookie fencing problem remains tracked by apache/bookkeeper#4812, and non-empty ledgers closed at a lower LAC are outside this PR’s scope.

Could you please take another look?