[fix][ml] Fail the read explicitly when a delivered batch contains a null slot

[dao-jun]

Motivation

The size loop in OpReadEntry.internalReadEntriesComplete dereferences every slot of the delivered batch:

for (int i = 0; i < entriesCount; i++) {
    entriesSize += returnedEntries.get(i).getLength();
}

A batch containing a null slot — a mixed range-cache read leaves its slot null when it drops an entry positioned outside the requested range — currently surfaces as a bare NullPointerException:

java.lang.NullPointerException: Cannot invoke "org.apache.bookkeeper.mledger.Entry.getLength()"
    because the return value of "java.util.List.get(int)" is null

which is then funneled through the exception fallback (Fallback to readEntriesFailed for exception in readEntriesComplete). Two problems with that shape:

1. The consumer sees an opaque NPE-wrapped failure with no indication of what went wrong.
2. The fallback path never releases the delivered entries — the whole batch's buffers leak (the valid siblings of the null slot included).

Modifications

• Detect a null slot up front in internalReadEntriesComplete, release the batch (null-safe), and fail the read with a proper ManagedLedgerException naming the slot index — the existing fallback forwards it as a normal read failure.
• Add a regression test delivering a batch with one null slot: it asserts the explicit failure message and a zero refcount on the valid sibling. Without the guard the test reproduces both the original NPE and the leak.

Verifying this change

• ./gradlew :managed-ledger:test --tests "org.apache.bookkeeper.mledger.impl.OpReadEntryNullSlotTest" — passes with the guard; removing the guard turns the test red with the original NPE.

[void-ptr974]

Thanks for adding this safeguard. There is one execution path where this check is reached too late.

With the in-flight read limiter enabled, RangeEntryCacheImpl.doAsyncReadEntriesWithAcquiredPermits wraps the callback. Its readEntriesComplete method iterates every entry and calls ((EntryImpl) entry).onDeallocate(...) before invoking the original OpReadEntry callback.

Therefore, if the assembled result is [validEntry, null], the wrapper throws an NPE first and OpReadEntry.internalReadEntriesComplete is never called. The valid entries and the acquired limiter permit can remain retained, and the read callback is not completed by this path.

The existing regression test covers the direct OpReadEntry behavior. An additional case going through RangeEntryCacheImpl with the in-flight limiter enabled would help cover this callback path as well.

[dao-jun]

Thanks for adding this safeguard. There is one execution path where this check is reached too late.

With the in-flight read limiter enabled, RangeEntryCacheImpl.doAsyncReadEntriesWithAcquiredPermits wraps the callback. Its readEntriesComplete method iterates every entry and calls ((EntryImpl) entry).onDeallocate(...) before invoking the original OpReadEntry callback.

Therefore, if the assembled result is [validEntry, null], the wrapper throws an NPE first and OpReadEntry.internalReadEntriesComplete is never called. The valid entries and the acquired limiter permit can remain retained, and the read callback is not completed by this path.

The existing regression test covers the direct OpReadEntry behavior. An additional case going through RangeEntryCacheImpl with the in-flight limiter enabled would help cover this callback path as well.

Good point, I'll check it

[dao-jun]

Good catch — verified: with the limiter enabled the wrapper's readEntriesComplete dereferences every slot before the original callback runs, so a [validEntry, null] batch NPEs there first, and neither the callback completion nor the permit release ever happens.

Pushed a2f3a9d. The wrapper now treats a null slot as a share of the permits with no entry that will ever return it: releasePermits runs immediately for that slot, and the batch is still delivered to the original callback, where the OpReadEntry null-slot guard fails the read explicitly and releases the valid entries (their hooks then return the remaining permits).

The new regression test drives a mixed read through RangeEntryCacheImpl with the in-flight limiter enabled — the storage leg returns a short result so the assembled batch carries a null — and asserts both the delivery and the full permit return (it times out without the wrapper fix).

[void-ptr974]

Thanks for adding the limiter coverage. It may also be helpful to cover the timeout path: when read timeout is enabled and the timeout completes before a sparse result arrives, ManagedLedgerImpl.ReadEntryCallbackWrapper handles the late success before OpReadEntry. Its late-result cleanup uses returnedEntries.forEach(Entry::release), which stops at a null slot; entries after the null may remain retained, and their deallocation hooks would not return the limiter handle.

The new limiter test exercises RangeEntryCacheImpl directly with a raw callback. An additional end-to-end case through RangeEntryCacheImpl -> limiter wrapper -> ReadEntryCallbackWrapper -> OpReadEntry could let the timeout complete first, then deliver [valid, null, valid], and verify that both valid entries are released and the limiter capacity is fully restored.