[improve][ml] Optimize individual-ack filtering by skipping scans for batches without acked entries

[void-ptr974]

Motivation

Before this change, ManagedCursorImpl.filterReadEntries used the global span of individually acknowledged positions to decide whether a read batch needed per-entry filtering.

A span records only the outer boundaries and does not represent holes between acknowledged positions or ledgers. As a result, a batch entirely inside one of those holes could still appear connected to the span. The cursor would then check every entry and allocate a new result list even though the batch contained no individually acknowledged entries.

Modifications

• Add PositionRangeSet.containsAny to query whether a ledger-local entry range contains an individually acknowledged position.
• Document its closed-range contract and pin negative lower-bound handling with regression coverage.
• Use the exact bitmap range query for ordered, single-ledger batches before performing per-entry filtering.
• Document the ordered, non-empty input contract and conservatively fall back to per-entry filtering for unexpected cross-ledger or descending batches.
• Read endpoint IDs directly and remove the eager entriesRange allocation from the read path.
• Add deterministic boundary tests and 96,000 reproducible randomized comparisons against a boolean-array oracle.
• Add a managed-cursor regression test for the span-hole fast path and defensive tests for the cross-ledger and descending fallbacks, including entry release behavior.
• Add a JMH benchmark for hit, no-hit, sparse-gap, missing-ledger, and empty-set scenarios.

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.PositionRangeSetTest \
  --tests org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testReadEntriesWithSkipDeletedEntries \
  --tests org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testFilterReadEntriesSkipsFilteringForGapBetweenIndividualAcks \
  --tests org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testFilterReadEntriesFallsBackForCrossLedgerBatch \
  --tests org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testFilterReadEntriesFallsBackForDescendingBatch \
  -PtestRetryCount=0
./gradlew :microbench:compileJava :microbench:checkstyleMain quickCheck

Representative JMH results on Java 21 with a batch size of 100 are shown below. These figures compare oldSpanDecision with newContainsAnyDecision. Their batch inputs are prepared in @Setup, so the results isolate the decision logic rather than end-to-end read-path latency:

Scenario	Span-based check	Bitmap range check
Sparse gap	4132 ns/op	26 ns/op
Missing ledger	1494 ns/op	5 ns/op
Outside span	79 ns/op	23 ns/op

The sparse-gap and missing-ledger rows represent false positives from the old global-span check. The exact ledger-local lookup lets those batches avoid unnecessary per-entry membership checks and intermediate result-list allocation. Cases containing acknowledged entries continue to use the existing per-entry filtering path. Additionally, filterReadEntries now logs the endpoint IDs directly instead of eagerly constructing an entriesRange.

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]

Thanks for this — it arrives with more rigor than a managed-ledger performance change usually does: a randomized oracle test against a boolean array, deterministic boundary cases, a regression test at the cursor call site, and a JMH harness across eight scenarios that self-checks old-vs-new agreement in @Setup. The optimization is sound and the reasoning in the description holds up.

I went over it with AI assistance (Claude + Codex gpt-5.6-sol) and then verified the findings myself locally at f4d9668, including mutation testing. I found no correctness or thread-safety issues.

Within a single ledger the new check is exact rather than approximate: for 0 <= e <= Integer.MAX_VALUE, contains(L, e) reduces to bitmap.contains(e) because getSafeEntry(e) == (int) Math.max(e, -1) == e, so containsAny(L, first, last) is precisely ∃ e ∈ [first, last] : contains(L, e). Combined with the batch precondition I note in the comment on the new condition (every entry in a batch lies in [first, last] of a single ledger), no acknowledged entry can leak back to a consumer.

Read-position advancement is unaffected as well: the batches that newly take the fast path are exactly the ones where the old path filtered nothing, so entriesCount != filteredEntries.size() in OpReadEntry.internalReadEntriesComplete evaluates the same either way and nexReadPosition is unchanged.

Verified locally at f4d9668:

• testFilterReadEntriesSkipsFilteringForGapBetweenIndividualAcks genuinely pins the change — reverting filterReadEntries to the old span-based check makes it fail on the isSameAs assertion. Good test.
• PositionRangeSetTest is green, and testContainsAnyRandomizedAgainstBooleanOracle costs 0.33 s, so the 96,000 comparisons are not a CI cost concern.
• :microbench:compileJava, :microbench:checkstyleMain, :managed-ledger:checkstyleMain and :managed-ledger:checkstyleTest all pass.

I'm marking this request-changes for one item only; everything else below is non-blocking, and I'm happy to flip to approve as soon as it lands.

The Math.max(0, lowerEntryId) guard in containsAny is defensive but unpinned — I verified by mutation that deleting it leaves the entire test class green. It is latent today (no caller can pass a negative entry id), so this is not an active bug. But it is exactly the kind of guard a later "simplification" removes silently, and the failure mode at the filterReadEntries call site would be redelivering acknowledged messages. One assertion closes it, and there is a one-click suggestion inline on testContainsAnyBoundaries.

Five inline comments in total: one I would like addressed (the clamp on PositionRangeSet.java, with the exact fix suggested on testContainsAnyBoundaries), and three non-blocking — a Javadoc nit on containsAny, a precondition worth writing down at the call site, and an optional allocation cleanup.

[void-ptr974]

The SplitManagerTest failure observed in this PR exposed an in-flight cleanup ordering race in SplitManager and UnloadManager. #26363 fixes the production cleanup ordering and adds deterministic regression tests for timeout, completion, retry, and concurrent cleanup paths.

[lhotari]

Re-reviewed at ff214557efe. All five round-1 comments are addressed, and I checked each against the code rather than the replies. Approving — the single item below is non-blocking.

What I verified

The Math.max(0, lowerEntryId) clamp is now genuinely pinned. That was my only blocking ask, so I re-ran the exact round-1 mutation: substituting bitmap.nextPresentValue(lowerEntryId) now fails PositionRangeSetTest.testContainsAnyBoundaries at line 268 — the new assertTrue(set.containsAny(1, -1, 0)). As I noted when suggesting it, line 269 passes either way; it documents the empty-range contract rather than pinning the clamp. Closed.

The containsAny Javadoc is accurate, not just present. Each claim checks out against the implementation: cardinality really is inclusive/inclusive (LongPairRangeSet: "from lower (inclusive) to upper (inclusive)", implemented as rank(upper+1) - rank(lower)), addOpenClosed really does lowerEntryIdOpen + 1, and nextPresentValue really returns -1 for from < 0 — which is exactly what makes the clamp sentence true. Both {@link}s resolve; javadoc -Xdoclint:all,-missing on the file exits 0.

The new firstEntryId > lastEntryId arm is a strict improvement, and it's worth writing down why, because "conservative fallback" undersells it. On master a descending batch never reached a conservative path at all: Range.closed(first, last) threw IllegalArgumentException: Invalid range: [1:10..1:1] (Position.compareTo orders by ledger then entry), which OpReadEntry converted into readEntriesFailed — the whole read failed. Degrading to per-entry filtering returns the correct result instead. It is pinned by testFilterReadEntriesFallsBackForDescendingBatch.

The fast path no longer touches Entry.getPosition() — the lazy Position allocation is gone from it along with the Range and its two Cuts. The four new .attr(...) calls bind to slog's primitive attr(String, long) overload, so removing the Range didn't reintroduce boxing.

No compatibility surface moves. containsAny is package-private on the package-private PositionRangeSet and correctly stays off the public LongPairRangeSet. The fast path returning the caller's own list was already reachable pre-PR, and OpReadEntry only reads it. quickCheck is green (checkstyle and rat both executed); PositionRangeSetTest plus the three filterReadEntries tests are green on a clean tree — 26 tests, 0 failures.

CI

The only red job is SplitManagerTest.testTimeout in Broker Group 1 — a load-balancer in-flight cleanup race with nothing to do with managed-ledger, which you've already root-caused in #26363. Re-running the failed job now.

[lhotari]

Non-blocking, and the only thing left from my side.

This test doesn't pin the guard it's named after. The batch is 1:10 → 2:0 → 2:1, so firstEntryId (10) > lastEntryId (1) is also true — the descending arm fires on this input and the cross-ledger arm never gets to matter.

Mutation-checked both ways on a clean tree:

mutation	testFilterReadEntries*
delete firstLedgerId != lastLedgerId	3/3 green — survives
delete firstEntryId > lastEntryId	…FallsBackForDescendingBatch fails, as it should

So the descending guard is pinned and the cross-ledger one isn't — the same shape of gap as the clamp in round 1, and the same risk: someone later deletes the ledger comparison as "redundant" and nothing goes red.

One character fixes it — make the batch ascending across the ledger boundary so only the ledger comparison can force the fallback:

Suggested change

	Position firstPosition = PositionFactory.create(1, 10);
	Position firstPosition = PositionFactory.create(1, 0);

With 1:0 → 2:0 → 2:1, 0 > 1 is false and containsAny(1, 0, 1) finds nothing in ledger 1 (the ack is at 2:0), so the ledger guard becomes the only thing standing between the batch and a skipped filter. Verified: green against the real code, and it fails the mutant that today survives. It also fits the test's name and comment better than a descending batch does.