[fix][ml] Skip contiguous deleted ranges during reads

[coderzc]

Motivation

When an older unacknowledged entry pins a cursor's mark-delete position, acknowledgments for later entries can form a very large contiguous individually deleted range. This is common for Key_Shared subscriptions when consumers continue acknowledging messages after the blocking entry.

ManagedLedgerImpl.internalReadFromLedger pre-filters a scan window before reading it. When every position in the window is skipped, it currently advances only to the position after that window. Reading past a large individually deleted range therefore takes O(entries / batchSize) read-loop iterations instead of one range lookup.

For closed ledgers, each iteration also repeats the fully-acknowledged ledger check. That check calls getNumberOfEntries and RangeSetWrapper.cardinality, which clones the cursor's per-ledger Roaring bitmap. Repeating it while walking a large acknowledged range can stall delivery and consume significant broker CPU.

Modifications

• When a complete scan window is skipped, use ManagedCursorImpl.getNextAvailablePosition to hop to the end of the containing individually deleted range.
• Preserve the existing next-position behavior when no cursor is associated with the read or when the window was skipped for a reason other than individual acknowledgment.
• Add a regression test with a closed ledger, a pinned mark-delete position, and a 19,996-entry contiguous individually deleted range. The test verifies both the returned entries and that the reader examines only a bounded number of positions.

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 --tests org.apache.bookkeeper.mledger.impl.ManagedCursorSkipDeletedEntriesTest --no-daemon
• ./gradlew :managed-ledger:test --tests 'org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testReadEntriesWithSkipDeletedEntriesAndWithSkipConditions' --no-daemon
• ./gradlew quickCheck --no-daemon

The regression test examines 20,002 positions and fails on the unpatched implementation. With this change, the test passes by hopping over the deleted range.

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

[lhotari]

LGTM — the change is correct and I'd merge it.

I verified the regression test empirically, building the module in worktrees at both revisions:

• PR head (4971f56): passes, 1 test, 26.6s.
• master with only the new test file copied in: fails — read examined 20002 positions to return 5 entries over a deleted range of 19996 entries.

So it is a genuine regression test, and the counts match a hand-trace of the read loop (~105 examined positions with the fix — 2 + 99 + 3 + 1 across four scan windows — versus 20,002 without).

I also put the diff through a multi-model review (Claude Fable and Codex gpt-5.6-sol reviewing independently with full repo context, then cross-validating each other's findings). Everything that survived is documentation or test polish — none of it blocks this PR.

Non-blocking findings

1. The new test's javadoc cites a class that no longer exists on master — ManagedCursorSkipDeletedEntriesTest.java:51-53 (and the same claim in the PR description) says the repeated check goes through RangeSetWrapper#cardinality, "which clones the cursor's per-ledger Roaring bitmap". There is no RangeSetWrapper in the tree. The actual path is isLedgerFullyAcked → getNumberOfEntries → PositionRangeSet.cardinality (PositionRangeSet.java:297-315) → LongBitmap.rank, and ConcurrentRoaringBitmap.rank (:273-286) takes a stamped read lock around RoaringBitmap.rank — no clone (cloning only happens in forEachLong/serialize). The repeated per-iteration cost is entirely real — executor round-trip + locked rank-based cardinality + a full window walk — just not a bitmap clone. Worth rewording in the javadoc and the description together, since the javadoc ships with the code. The wording looks like it was carried over from branch-4.x.

2. The comment overstates the fallback — ManagedLedgerImpl.java:2429-2430. "falls back to the next position when the window was skipped for any other reason" — the real condition is narrower: it falls back only when lastEntry itself is not inside an individually deleted range. A mixed window (caller-skipped at the head, a deleted run through lastEntry) still hops, which is correct behaviour but not what the comment describes. Also suggest "already individually acknowledged": positions covered by mark-delete are removed from individualDeletedMessages via removeAtMost, so they never hop.

3. The opReadEntry.cursor != null guard is unreachable — ManagedLedgerImpl.java:2431. OpReadEntry.create dereferences cursor at OpReadEntry.java:66 before assigning any field, so a null cursor never produces an op; updateReadPosition → cursor.setReadPosition dereferences it on the very next line, as does asyncReadEntry via opReadEntry.cursor::getNumberOfCursorsAtSamePositionOrBefore (ManagedLedgerImpl.java:2467). The other creator, OpScan, passes skipCondition = null and never enters this branch. Harmless, but the ternary and the "when no cursor is associated with the read" sentence describe a state that cannot occur — consider dropping both.

4. Test runtime is on the heavy side — 26.6s for a single regression. Dropping TOTAL_ENTRIES to ~1,000 preserves the discrimination (unpatched would examine ~1,002 against the <= 400 threshold; patched ~106). Entirely optional — reviewers split on whether 26.6s is out of line for this module. No groups annotation is needed: ungrouped tests run in the managed-ledger job, and the test did execute.

5. Informational, no change recommended — the examinedPositions counter only observes every position because asyncReadEntriesWithSkip composes skipCondition.or(this::isMessageDeleted) and Predicate.or short-circuits left to right (the test comment already documents this). Two hardening guards were considered and both rejected: assertTrue(examinedPositions.get() > 0) would pass vacuously if the composition were ever flipped, since deliverable positions are still counted; > read.size() is too tight to stay stable. Leaving it as-is is the right call.

One optional code comment worth adding

The hop can now advance the cursor read position well past opReadEntry.maxPosition, where the previous code overshot by exactly one entry. This is safe, but subtly enough that a line of comment would help: every position crossed is individually deleted at lookup time under the cursor read lock, so nothing deliverable or transaction-undecided is jumped, and both the readPosition > maxPosition early return (ManagedLedgerImpl.java:2355-2358) and checkReadCompletion terminate the op cleanly.

Worth noting that the tempting justification — "acknowledged implies delivered implies ≤ maxReadPosition" — is actually false: under autoSkipNonRecoverableData, OpReadEntry.internalReadEntriesFailed computes its skip span via getValidPositionAfterSkippedEntries (ManagedLedgerImpl.java:4320-4334), which never consults maxPosition, and skipNonRecoverableEntries (ManagedCursorImpl.java:3165-3205) then individually deletes every entry in that span, including never-delivered ones. The sound argument is that the crossed ground is deleted, not that it was delivered.

On hopping earlier in the scan

Since it will probably come up: it is possible to hop before the per-position skip walk, but it is not worth doing.

The trap is that getNextAvailablePosition(p) returns p.getNext() whenever p is not individually deleted (ManagedCursorImpl.java:3972), so it advances on a miss. The current placement gets its precondition for free — reaching firstValidEntry == -1 proves every position through lastEntry was tested and skipped. A pre-check has no such proof, and a naive firstEntry = hop.getEntryId() would silently skip the first live entry of every read whose window starts on a live entry.

Even with a correct guard the payoff is a bounded constant: this PR already collapses the crossing from O(range/batch) read-loop iterations (each an executor round-trip plus a closed-ledger isLedgerFullyAcked) down to one, and an earlier hop would only remove the single remaining ≤batchSize predicate walk — while adding a rangeContaining lookup (cursor read lock + fastutil RB-tree get + bitmap contains) to skip processing that gains nothing on delayed-delivery-heavy topics. The post-scan placement in this PR is the better anchor: one iteration per range, it reuses knowledge the scan already produced, and it costs nothing on the happy path.

Checked and cleared

For the record, these were examined and found not to be problems: cross-ledger overshoot (PositionRangeSet.rangeContaining is per-ledger and Position.getNext() stays in-ledger, so the target is always (sameLedger, e+1)); overshooting LAC; infinite loop or stall (both branches of getNextAvailablePosition return strictly greater than the input); partially-acked batch entries (they only enter individualDeletedMessages once the bitset empties, ManagedCursorImpl.java:2637-2644); hopping over delayed-delivery entries (delayed state is not represented in individualDeletedMessages); and OpScan, which passes a null skipCondition and never reaches this branch — which is required, since scans must see deleted entries.

Review assisted by Claude Fable and Codex gpt-5.6-sol; findings verified against the code and the test results reported above by me.