[improve][broker] Reduce persistent topic backlog check overhead with ManagedCursor.hasBacklog

[void-ptr974]

Motivation

Fixes #26036.

Some persistent topic paths only need to know whether backlog exists, but currently call
getNumberOfEntriesInBacklog(...) and compare the returned count with zero. For precise checks, this
can add unnecessary overhead because the caller does not need the exact backlog count. For imprecise
checks, using a boolean API also better expresses the intent and keeps the call sites consistent.

Modifications

• Add backward-compatible hasBacklog methods to ManagedCursor, Subscription, and Replicator.
• Implement ManagedCursorImpl.hasBacklog() to short-circuit once it finds one readable backlog entry.
• Preserve correctness for individually deleted messages by skipping deleted ranges.
• Override persistent subscription and replicator hasBacklog methods to delegate to the managed cursor.
• Replace backlog-count presence checks with hasBacklog(...) in persistent topic, subscription,
  dispatcher, compactor subscription, and replicator paths while preserving the existing precise or
  imprecise mode at each call site.
• Add managed-ledger tests that compare hasBacklog() / hasBacklog(true) /
  hasBacklog(false) with the existing backlog count APIs across empty, caught-up, mixed ack,
  cross-ledger, batch-index, reset, recovery, deleted-range-to-tail, negative-counter fallback, and
  uninitialized mark-delete states.
• Add deterministic generated operation coverage for mixed add, delete, mark-delete, reset, and reopen
  sequences.
• Add broker tests covering real producer/consumer acknowledgement orders, batching, and subscription
  lifecycle operations such as skip, reset, expire, and clear backlog.

Performance impact

This reduces persistent topic backlog check overhead by avoiding exact backlog counting when callers only
need a boolean result. In the common precise-check case, the check can stop at the first valid backlog entry.

For existing imprecise checks, the patch preserves the same imprecise mode and only replaces count-based
presence comparisons with the equivalent boolean API.

Verifications

• ./gradlew :managed-ledger:test --tests org.apache.bookkeeper.mledger.impl.ManagedCursorTest.hasBacklog*
• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.service.persistent.PersistentTopicTest.testHasBacklogTracksRealProduceConsumeAckOrders --tests org.apache.pulsar.broker.service.persistent.PersistentTopicTest.testHasBacklogTracksRealSubscriptionLifecycleOperations
• ./gradlew :managed-ledger:checkstyleMain :managed-ledger:checkstyleTest :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest

[lhotari]

good change, one comment

[lhotari]

Please search for all usages of cursor.getNumberOfEntriesInBacklog(boolean) where the result is compared against 0. There seems to be some more usages that need to be addressed.

[void-ptr974]

Good catch. I updated the remaining backlog presence checks to use hasBacklog(...) while preserving the existing precise/imprecise modes, and added coverage for the updated paths. Please take another look when you have a chance.

[lhotari]

org.apache.pulsar.broker.service.AbstractReplicator#disconnect seems to have one remaining backlog check based on count.

[void-ptr974]

Good catch, missed that one. Changed.

[lhotari]

btw. I did some analysis with Claude Code about cases where there would be individualDeletedMessages that aren't already absorbed into the markDeletePosition.

In the normal ack flow it usually can't, because the delete path is greedy: asyncDelete (ManagedCursorImpl.java:2700-2715) absorbs any deleted range contiguous with the mark-delete
position into MD (newMarkDeletePosition = range.upperEndpoint(), then setAcknowledgedPosition → individualDeletedMessages.removeAtMost(markDeletePosition) at line 2261). That maintains
an invariant that getNextValidPosition(markDeletePosition) is a real, non-deleted entry whenever MD < last — under which mark-delete-only would agree with the walk.

The walk exists to not depend on that invariant:

• Recovery (recoverIndividualDeletedMessages, line 770) rebuilds the range set verbatim with raw addOpenClosed calls and does not re-run the absorption, including ranges synthesized
  across trimmed/empty ledgers (addOpenClosed(li, -1, …), line 798-802).
• The directly-constructed test state above is reachable precisely because it bypasses the public delete API.

I guess the backlog check in hasBacklog would short circuit in realistic cases with markDeletePosition.compareTo(lastPosition) >= 0 or the nextPosition.compareTo(lastPosition) > 0. This is good.

[lhotari]

LGTM, thank you @void-ptr974