[improve][ml] Support inflight reads limiter when managed ledger cache is disabled

[poorbarcode]

Motivation

managedLedgerMaxReadsInFlightSizeInMB limits memory reserved by reads only when the entry cache is enabled. With managedLedgerCacheSizeMB=0, EntryCacheDisabled previously bypassed the limiter, so a broker configured without an entry cache had no equivalent protection against excessive in-flight read memory.

Modifications

• Pass the existing shared in-flight reads limiter to EntryCacheDisabled, so cache-disabled reads are limited by managedLedgerMaxReadsInFlightSizeInMB.
• Keep managedLedgerMaxReadsInFlightSizeInMB=0 as the existing way to disable the limiter; no new broker setting is introduced.
• Release permits after entries are deallocated, and release on read failures.
• Preserve ManagedLedger executor affinity for callback delivery.
• Do not invalidate a healthy ledger handle for limiter or BookKeeper TooManyRequests backpressure.
• Add integration coverage for permit accounting, callback exceptions, read failures, queued reads, and timeout diagnostics.

Deployment

No configuration migration is required. Brokers that set a positive managedLedgerMaxReadsInFlightSizeInMB will also apply it to cache-disabled reads after upgrading.

[lhotari]

do we need a separate configuration parameter at all? it could be considered a bug that managedLedgerMaxReadsInFlightSizeInMB doesn't have effect when entry cache is disabled.

[poorbarcode]

Agreed. I removed the additional configuration parameter entirely. EntryCacheDisabled now always receives the existing broker-level limiter; managedLedgerMaxReadsInFlightSizeInMB remains the sole control, and an explicit value of 0 still disables limiting.

[lhotari]

Thanks for this — enabling MaxReadsInFlightSize on the cache-disabled path is genuinely worth doing, and the core design correctly mirrors the proven RangeEntryCacheImpl pattern: acquire-before-read with queue/timeout, release on last deallocate via EntryImpl.onDeallocate, immediate release on empty results and read failures, and correctly not releasing a success()==false handle. I traced every normal exit path and permits are accounted exactly once on all of them. The config plumbing is complete and consistent end to end, both conf files were updated, and the old public EntryCacheDisabled(ManagedLedgerImpl) constructor is retained so binary/source compatibility holds.

I also checked two things that could have been much worse and are fine: surfacing TooManyRequestsException from this path is safe (OpReadEntry treats it as transient at OpReadEntry.java:175, and the dispatcher/replicator readEntriesFailed handlers all retry with backoff), and the release-from-arbitrary-thread pattern respects the limiter's locking contract.

The problems are all on the abnormal paths, plus one scope issue. Requesting changes primarily because of #1 and #2.

1. A throwing downstream callback double-releases the handle and permanently corrupts the limiter

readEntries chains .exceptionally(...) directly onto thenAcceptAsync(body, ...), and body ends with callback.readEntriesComplete(entries, ctx). So a Throwable escaping the caller's callback completes that stage exceptionally, and .exceptionally then invokes readEntriesFailed on the same wrapper — after the release hooks were already installed. Two releases for one acquire:

• empty-result branch: release(handle), then the caller throws, then release(handle) again — immediate and deterministic;
• non-empty branch: release(handle) from .exceptionally, plus a second one later when the entries deallocate.

InflightReadsLimiter.internalRelease does remainingBytes += handle.permits with no idempotency check and no clamp, so the drift is permanent: remainingBytes ends up above maxReadsInFlightSize, the buffer-size gauge goes negative, and — because the oversized-read escape hatch tests remainingBytes == maxReadsInFlightSize for exact equality — large reads can afterwards only ever exit via acquire timeout. The direct-memory protection this PR exists to add is silently weakened for the life of the broker, with no log line.

Worth noting RangeEntryCacheImpl does not have this shape: there the caller's callback runs inside whenComplete bodies (PendingReadsManager.java:237-245, 368-375), so a throwing readEntriesComplete cannot be re-routed into readEntriesFailed. This path is uniquely exposed.

Reachability is real rather than theoretical: OpReadEntry.readEntriesComplete catches Throwable internally (OpReadEntry.java:120-124), so the main cursor path is shielded — but the replay path is not. ManagedCursorImpl.asyncReplayEntries (:1915) reaches PersistentDispatcherMultipleConsumers.readEntriesComplete (:673), which has no top-level catch.

The cheapest robust fix is a consumed-flag or a clamp in InflightReadsLimiter.release so a second release is a no-op — worth doing regardless, since the limiter is one stray release away from corruption today. Separating success and failure delivery in readEntries (e.g. whenCompleteAsync, or moving the caller invocation out of the thenAcceptAsync body) would fix the root cause.

2. The single-entry rewrite is not behind the feature flag

asyncReadEntry(lh, position, ...) now delegates to the range overload unconditionally — the inflightReadsLimiter == null early-return happens inside the delegate, so the restructuring applies to every managedLedgerCacheSizeMB=0 broker on upgrade, flag or no flag. That contradicts "the default value is false to avoid break changes", and it ships to 4.0.14/4.2.5. Three distinct changes versus base:

(a) Double callback plus spurious handle invalidation. Base used whenCompleteAsync, which cannot fire twice. Now, if callback.readEntryComplete throws, .exceptionally fires readEntriesFailed, which calls ml.invalidateLedgerHandle(lh) and then readEntryFailed — i.e. a successful read reports both outcomes and tears down a healthy ledger handle. Previously impossible, and reachable with the flag off via admin peek/examine (asyncGetNthEntry), replay reads, and OpFindNewest.

(b) Failure callbacks change thread. Base pinned both outcomes to ml.getExecutor(). The new .exceptionally has no executor, so readEntryFailed now runs on whichever thread completed the BK future — or fully inline on the caller's stack, since ReadEntryUtils.readAsync returns an already-failed future in three cases (:41, :45, :49), including reads past the LAC. Callers written against the always-async, executor-confined contract now get reentrancy on their own thread.

(c) invalidateLedgerHandle is skipped for TooManyRequestsException (:174). Because createManagedLedgerException maps BK's own TooManyRequests code to that same class, this also suppresses invalidation for genuine bookie-side throttling. I think this is the right policy — it matches RangeEntryCacheImpl.readFromStorage:560-563 — but it's an unannounced semantic change on the default path.

If the intent really is "no behaviour change unless you opt in", the single-entry overload should either keep its whenCompleteAsync shape when the limiter is absent, or the PR description should state plainly what changes unconditionally.

3. Test covers only the happy path

The new test does prove the core claim well and deterministically — acquire before completion, held while entries are alive, released when they are — and it cleans up in finally. But every branch where this feature can go wrong is unexercised: the deferred-acquire executor hop (the only cross-thread hand-off in the new code), queue-full TooManyRequestsException, acquire timeout, release-on-read-failure, the flag-off gating that the whole backport argument rests on, the rewritten single-entry overload, and concurrent reads.

assertTrue(limiter.getRemainingBytes() < totalCapacity) also never pins how many permits, so the new getEstimatedEntrySize arithmetic is functionally unverified — the pre-existing testPreciseLimitation shows exact-value assertions are practical here. A permit leak is this feature's worst failure mode and would currently ship green.

Smaller items

• Permit leak on executor rejection (:87-89). InflightReadsLimiter.handleQueuedHandle deducts permits before invoking the callback and only logs exceptions from it, so a RejectedExecutionException from ml.getExecutor().execute leaks the permits and drops the read entirely. The same unguarded shape exists on master in RangeEntryCacheImpl, so this is a new occurrence of a known pattern rather than a new defect — mostly shutdown-only in practice.
• No try/catch around the synchronous read start. RangeEntryCacheImpl wraps both asyncReadEntry overloads in try { … } catch (Throwable t) { callback.readEntriesFailed(...) }; this one doesn't, so a synchronous throw after acquire leaks the full estimate and hangs the read.
• getEstimatedEntrySize is a fork of RangeEntryCacheImpl.getEstimatedEntrySize, minus the Math.max(getAvgEntrySize(), …) refinement (justifiable here — no cache stats — but undocumented). The non-empty branch is character-identical; a shared static helper would stop the two copies drifting. The arithmetic itself checks out: long throughout, Math.max(1, …) guards the division, and the getLastAddConfirmed() < 0 guard handles in-progress ledgers.
• The TooManyRequestsException message has no diagnostic context — compare RangeEntryCacheImpl.java:345-351, which includes ledger id, ML name, estimated size, entry count, and the tunable config keys. Both queue-full and timeout failures funnel through the one terse string.
• Release guard uses == 0 at :111 where RangeEntryCacheImpl:364 defensively uses <= 0.
• Config name and checklist. enableReadsInFlightLimiterEvenIfManagedLedgerCacheDisabled sorts away from the managedLedger* family it modifies (managedLedgerMaxReadsInFlightSizeInMB, …AcquireTimeoutMillis, …AcquireQueueSize). Since you plan to flip the default later, the name is long-lived and there's no rename path once it ships on three branches — something like managedLedgerReadsInFlightLimiterEnabledWhenCacheDisabled would fit. The "default values of configurations" and "affects deployment" checkboxes are also unchecked despite adding a key to both conf files.
• The flag is inert unless managedLedgerMaxReadsInFlightSizeInMB > 0, which the doc doesn't say. On master that's masked because ManagedLedgerClientFactory substitutes 15% of direct memory when unset — but on branch-4.0 and branch-4.2 it defaults to 0, so an operator on 4.0.14/4.2.5 could set this flag, restart, and get no limiting at all with nothing explaining why. Worth stating in the doc, or warning at startup when the flag is true while the limiter is disabled.

Backport

Beyond item 2: this won't cherry-pick cleanly to branch-4.0 — EntryCacheDisabled there takes boolean isSlowestReader rather than IntSupplier expectedReadCount and uses a different entry factory, so the adapter has to pick an isSlowestReader value by judgement. branch-4.2 is close to a clean pick. Given that the trickiest part of the patch is the permit accounting, I'd suggest opening the 4.0 backport as its own reviewed PR and letting this soak on master first.

[lhotari]

This release, plus the ones registered in readEntriesComplete above, can both run for a single acquire.

readEntries chains .exceptionally(...) onto thenAcceptAsync(body, ...), and body ends with callback.readEntriesComplete(...). So a Throwable from the caller's callback completes that stage exceptionally and lands here — after the onDeallocate hooks were already installed (or, in the empty-result branch, after release already ran at line 106).

InflightReadsLimiter.internalRelease does remainingBytes += handle.permits with no idempotency check and no clamp, so the over-credit is permanent: the buffer-size gauge goes negative and, because the oversized-read escape hatch tests remainingBytes == maxReadsInFlightSize for exact equality, large reads can afterwards only exit via acquire timeout.

RangeEntryCacheImpl isn't exposed to this — there the caller's callback runs inside whenComplete bodies (PendingReadsManager.java:237-245, 368-375), so a throwing readEntriesComplete can't be re-routed into readEntriesFailed.

Reachable in practice: OpReadEntry.readEntriesComplete catches Throwable (OpReadEntry.java:120-124) so the main cursor path is shielded, but ManagedCursorImpl.asyncReplayEntries (:1915) → PersistentDispatcherMultipleConsumers.readEntriesComplete (:673) has no top-level catch.

Cheapest robust fix is making release idempotent (a consumed flag, or clamping to maxReadsInFlightSize) — worth doing regardless. Fixing the root cause means not invoking the caller from inside the thenAcceptAsync body.

[poorbarcode]

Fixed. readEntries now uses thenApplyAsync only for conversion and accounting, followed by whenCompleteAsync for delivery. A throwing downstream success callback therefore cannot be routed to readEntriesFailed or release the same handle twice. Added a regression test that throws from readEntriesComplete and verifies both the limiter capacity and callback count.

[lhotari]

Two gaps around the acquire, both of which RangeEntryCacheImpl guards and this doesn't:

1. InflightReadsLimiter.handleQueuedHandle deducts the permits before invoking this callback, and wraps the invocation in a try/catch that only logs. So a RejectedExecutionException from ml.getExecutor().execute(...) (executor shutting down while reads are queued) leaks the permits and drops the read — the caller's callback never fires. The same unguarded shape exists on master in RangeEntryCacheImpl, so this is a new occurrence of a known pattern rather than a new defect, and it's mostly shutdown-only.

2. There's no try/catch around the synchronous portion at all. RangeEntryCacheImpl wraps both overloads in try { … } catch (Throwable t) { …; callback.readEntriesFailed(...) } (:236-265, :270-284). Here a synchronous throw out of readEntries/ReadEntryUtils.readAsync after permits are acquired leaks the full estimate and hangs the read.

[poorbarcode]

The ManagedLedger OrderedScheduler is created without maxTasksInQueue, whose BookKeeper builder default is unbounded. Executor rejection is therefore not a queue-saturation path; it can occur only during executor shutdown. Since the broker and its process-local limiter are being torn down together, a retained permit in this terminal race has no operational impact. The same executor hop already exists in RangeEntryCacheImpl, so I am keeping shutdown-path hardening out of scope here. ReadEntryUtils reports validation failures through failed futures; handling arbitrary synchronous failures would likewise be broader hardening outside this focused change.

[lhotari]

This message has no diagnostic context. The equivalent on the cache-enabled path (RangeEntryCacheImpl.java:345-351) includes the ledger id, the ML name, the estimated read size, the entry count, and the names of the tunable config keys.

Both the queue-full and the acquire-timeout failures funnel through this one string, so an operator hitting throttling on the cache-disabled path gets no topic/ledger/size information and no pointer to managedLedgerMaxReadsInFlight*. Since this exception is the main operator-visible symptom the feature introduces, it's worth matching the other path's detail.

[poorbarcode]

Fixed. The cache-disabled limiter failure now matches RangeEntryCacheImpl diagnostic context: ledger id, managed ledger name, estimated read size, entry count, and the related queue-size, timeout, and limit settings. Added an acquire-timeout integration test that asserts this context is returned.

[lhotari]

This delegation is unconditional — the inflightReadsLimiter == null early-return lives inside the delegate — so the single-entry path is restructured for every managedLedgerCacheSizeMB=0 broker on upgrade, whether or not the new flag is set. That conflicts with "the default value is false to avoid break changes", and it ships to 4.0.14/4.2.5.

Two consequences beyond the one flagged below:

• Double callback. Base used whenCompleteAsync, which cannot fire twice. Now a throwing callback.readEntryComplete reaches .exceptionally → readEntriesFailed → ml.invalidateLedgerHandle(lh) → readEntryFailed, so a successful read reports both outcomes and tears down a healthy ledger handle. Reachable with the flag off via admin peek/examine (asyncGetNthEntry), replay reads, and OpFindNewest.
• Failure callbacks change thread. Base pinned both outcomes to ml.getExecutor(). The new .exceptionally has no executor, so readEntryFailed runs on whichever thread completed the BK future — or inline on the caller's stack, since ReadEntryUtils.readAsync returns an already-failed future for reads past the LAC (:41, :45, :49). Callers written against the always-async, executor-confined contract now get reentrancy.

If the intent is "no behaviour change unless you opt in", could this overload keep its whenCompleteAsync shape when the limiter is absent?

[poorbarcode]

I removed the opt-in flag, so the default-false compatibility claim no longer applies. The double-callback issue is fixed by separating conversion from callback delivery, and callback delivery now explicitly uses the ManagedLedger executor for both success and failure. The behavior change is intentional: cache-disabled reads now participate in the same in-flight read memory protection as RangeEntryCacheImpl.

[lhotari]

This is the right policy — it matches RangeEntryCacheImpl.readFromStorage:560-563, and throttling isn't a reason to tear down a healthy ledger handle.

Worth calling out explicitly in the PR description though, because it also changes behaviour when the feature is off: createManagedLedgerException maps BK's own TooManyRequests code to this same class (ManagedLedgerImpl.java:4636-4638), so bookie-side throttling no longer invalidates the handle either, where base always did. Small and benign, but it's a semantic change shipping to two maintenance branches under a "default false, no break changes" claim.

[poorbarcode]

This is intentional. A limiter rejection happens before any BookKeeper read, so the handle is necessarily healthy. BookKeeper TooManyRequests is also transient backpressure rather than evidence of a bad handle; invalidating and reopening it does not relieve throttling and adds churn. This aligns the cache-disabled path with the existing RangeEntryCacheImpl policy.

[lhotari]

This is a copy of RangeEntryCacheImpl.getEstimatedEntrySize (:508-514) minus the Math.max(getAvgEntrySize(), …) refinement in the empty-ledger branch. Dropping that is justifiable here (no cache stats to derive an average from) but it's undocumented, and the non-empty branch is character-identical duplication.

The arithmetic itself checks out — long throughout, Math.max(1, …) guards the division, and the getLastAddConfirmed() < 0 guard correctly covers in-progress/empty ledgers. The concern is drift: a future fix to the estimator will land in one copy and silently miss the other, on branches where that's hardest to notice. A shared static helper would also let the new test assert exact permit amounts the way testPreciseLimitation does.

[poorbarcode]

The cache-disabled path has no cache history from which to derive the empty-ledger average, so it intentionally uses the existing 10 KiB fallback. The remaining handle-based calculation is small and local. I agree a shared estimator could reduce future drift, but it would be a refactoring beyond the cache-disabled limiter fix and is left for follow-up.

[lhotari]

Two things on the config:

Naming. Every sibling in CATEGORY_STORAGE_ML is prefixed managedLedger* (managedLedgerMaxReadsInFlightSizeInMB, …AcquireTimeoutMillis, …AcquireQueueSize), so this key sorts away from the family it modifies. Since you plan to flip the default to true later, the name is long-lived and there's no rename path once it ships on master + 4.2 + 4.0 — something like managedLedgerReadsInFlightLimiterEnabledWhenCacheDisabled would fit the convention.

Missing precondition in the doc. The doc names only managedLedgerCacheSizeMB=0, but the flag is equally inert unless managedLedgerMaxReadsInFlightSizeInMB > 0 (InflightReadsLimiter sets enabled=false when maxReadsInFlightSize <= 0). On master that's masked because ManagedLedgerClientFactory substitutes 15% of direct memory when unset — but on branch-4.0/branch-4.2 it defaults to 0, so an operator on 4.0.14/4.2.5 can set this flag, restart, and get no limiting at all with nothing explaining why. Worth stating in the doc, or logging a warning at startup when the flag is true while the limiter is disabled.

(Also: the PR's "default values of configurations" and "affects deployment" checkboxes are unchecked despite adding a key to both conf files.)

[poorbarcode]

Resolved by removing the new flag and its configuration entries. There is no naming or inactive-flag precondition now: cache-disabled reads use the existing managedLedgerMaxReadsInFlightSizeInMB setting directly, and a value of 0 disables the limiter. I also updated the PR description and deployment checklist.

[lhotari]

assertTrue(getRemainingBytes() < totalCapacity) never pins how many permits were taken, so the new getEstimatedEntrySize arithmetic and the (lastEntry - firstEntry + 1) * size multiplication are functionally unverified — any wrong estimate still passes. The pre-existing testPreciseLimitation in this same class shows an exact-value assertion is practical.

More broadly, this test proves the happy path well (acquire → held while entries live → released on release, with cleanup in finally), but every branch where the feature can go wrong is unexercised: the deferred-acquire executor hop (the only cross-thread hand-off in the new code), queue-full TooManyRequestsException, acquire timeout, release-on-read-failure, the flag-off gating that the backport argument rests on, the rewritten single-entry overload, and concurrent reads.

A permit leak is this feature's worst failure mode and would currently ship green — a release-on-failure test and a flag-off test feel like the minimum before merge.

[poorbarcode]

Expanded coverage substantially: the cache-disabled happy-path test now asserts the exact permit estimate; added regression coverage for a throwing downstream callback, read failure permit release, queued-read wakeup after Entry release, and acquire-timeout diagnostics. The flag-off case is no longer applicable because the additional flag was removed. Queue-capacity and concurrency behavior remain covered by the existing InflightReadsLimiter unit tests.