Skip to content

[improve][broker] Reduce allocations on dispatch and managed-ledger read hot paths - #26956

Open
Asthenia0412 wants to merge 4 commits into
apache:masterfrom
Asthenia0412:feat/pulsar-contrib
Open

Asthenia0412 wants to merge 4 commits into
apache:masterfrom
Asthenia0412:feat/pulsar-contrib

Conversation

@Asthenia0412

Copy link
Copy Markdown
Contributor

Background

Broker message dispatch and managed-ledger entry read paths allocate objects on every iteration, adding up to significant GC pressure at high throughput:

  1. In ManagedLedgerImpl.internalReadFromLedger, the skip-condition loop creates a new Position for every entry scanned, even though the most common skip predicate is cursor.isMessageDeleted which only needs (ledgerId, entryId).

  2. In AbstractBaseDispatcher.filterEntriesForConsumer, EntryBatchIndexesAcks stores batch-size and ack-set per entry using Pair<Integer, long[]>, forcing a Pair allocation for every partially-acked batch entry.

Both paths sit on the broker hot path between managed-ledger reads and consumer dispatch, so removing these per-iteration allocations reduces young-gen allocation rate and eases GC pressure.

Changes

managed-ledger read path

  • Add ManagedCursorImpl.isMessageDeleted(long, long) overload that avoids creating a Position by using Position.compareTo(long, long).
  • Add OpReadEntry.skipOnlyMessageDeleted boolean flag, set true when the original caller did not supply a custom skipCondition.
  • ManagedLedgerImpl.internalReadFromLedger: when skipOnlyMessageDeleted is true, call cursor.isMessageDeleted(ledgerId, entryId) directly instead of allocating PositionFactory.create() on every iteration.

broker dispatch path

  • Replace EntryBatchIndexesAcks.Pair<Integer, long[]>[] with two parallel arrays (int[] batchSizes, long[][] ackSets) so recording an entry ack info no longer allocates a Pair.
  • Add EntryBatchIndexesAcks.setIndexesAcks(int, int, long[]) as the new allocation-free API; keep the Pair-accepting overload as @deprecated.
  • AbstractBaseDispatcher.filterEntriesForConsumer: call the new overload.

Impact

At 100k entries dispatched per second, the broker avoids ~100k Pair alloc/s and the managed ledger avoids ~N_position alloc/s on its read skip loop, both removed from young-gen allocation rate. No functional change.

Testing

  • :managed-ledger:compileJava / :compileTestJava
  • :pulsar-broker:compileJava / :compileTestJava
  • :pulsar-broker:test EntryBatchIndexesAcksTest passed
  • :managed-ledger:test PositionRangeSetTest, ManagedCursorTest passed

@codelipenghui codelipenghui left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Requesting changes for the read-only cursor regression described inline.

Reproduced at head 30e786644bcc06ce6d5f44b60c5c19ece2ef591c: the existing org.apache.bookkeeper.mledger.impl.ReadOnlyCursorTest.simple test fails with ManagedLedgerException wrapping a NullPointerException because markDeletePosition is null. The stack goes through the new primitive deletion-check fast path. Please preserve subclass deletion-check semantics and verify the read-only cursor tests before merging.

@codelipenghui codelipenghui left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed the follow-up at e611a2defb9692fcc994461c66538901db40edfb. The read-only cursor override fixes the previously reported P1 regression by preserving the always-false deletion check on the primitive fast path.

All 8 tests in org.apache.bookkeeper.mledger.impl.ReadOnlyCursorTest passed, including simple, which failed at the previous head, and the asynchronous read test.

No actionable findings identified within the reviewed scope. Performance gains remain unmeasured.

@Asthenia0412

Copy link
Copy Markdown
Contributor Author

Hi @lhotari @codelipenghui - Thanks for the approval and thorough review! Would it be possible to merge this when you have a moment? No open threads remain and CI is green. Thanks!

@Asthenia0412

Copy link
Copy Markdown
Contributor Author

Hi @lhotari @codelipenghui - quick bump. This PR has maintainer approval, CI is green, and all review threads are resolved. Whenever you have a moment to merge, it would be much appreciated. Thanks!

@Asthenia0412

Copy link
Copy Markdown
Contributor Author

Rebased onto upstream/master (no conflicts). ReadOnlyCursor fix (e611a2d) — overriding isMessageDeleted(long, long) to preserve subclass semantics when the skip shortcut is active — remains in HEAD. Force-pushed updated branch for fresh CI. Thanks for your review!

@Asthenia0412

Copy link
Copy Markdown
Contributor Author

Hi @codelipenghui — I've resolved the read-only cursor regression with an override isMessageDeleted(long, long) on ReadOnlyCursorImpl that mirrors the always-false semantics of the existing Position variant. The ManagedLedgerImpl fast path at L2786 now correctly short-circuits before the deleted-entry check for read-only cursors (which never set markDeletePosition).

Pre-existing ReadOnlyCursorTest.simple passes locally against this head. CI is green. Could you re-review when you have a moment? Thank you! 🙏

Comment on lines +52 to +62
private static final int NUM_BOOKIES = 3;

@BeforeClass
void setup() throws Exception {
bkEnsemble = new LocalBookkeeperEnsemble(1, 0);
bkEnsemble = new LocalBookkeeperEnsemble(NUM_BOOKIES, 0);
bkEnsemble.start();
InternalConfigurationData internalConf = new InternalConfigurationData(
"127.0.0.1:" + bkEnsemble.getZookeeperPort(), null, "/ledgers", null, null);
URI dlogUri = WorkerUtils.initializeDlogNamespace(internalConf);
WorkerConfig workerConfig = new WorkerConfig();
workerConfig.setNumFunctionPackageReplicas(1);
workerConfig.setNumFunctionPackageReplicas(NUM_BOOKIES);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks like an unrelated change

Asthenia0412 and others added 4 commits October 20, 2026 10:16
…ead hot paths

## Background

Broker message dispatch and managed-ledger entry read paths allocate objects
on every iteration, adding up to significant GC pressure at high throughput:

1. In ManagedLedgerImpl.internalReadFromLedger, the skip-condition loop creates
   a new Position for every entry scanned, even though the most common skip
   predicate is cursor.isMessageDeleted which only needs (ledgerId, entryId).

2. In AbstractBaseDispatcher.filterEntriesForConsumer, EntryBatchIndexesAcks
   stores batch-size and ack-set per entry using Pair<Integer, long[]>, forcing
   a Pair allocation for every partially-acked batch entry.

## Changes

### managed-ledger read path
- Add ManagedCursorImpl.isMessageDeleted(long, long) overload that avoids
  creating a Position by using Position.compareTo(long, long) (already
  present since the Position interface).
- Add OpReadEntry.skipOnlyMessageDeleted boolean flag, set true when the
  original caller did not supply a custom skipCondition, so the managed-ledger
  read loop can take the allocation-free fast path.
- ManagedLedgerImpl.internalReadFromLedger: when skipOnlyMessageDeleted is true,
  call cursor.isMessageDeleted(ledgerId, entryId) directly instead of allocating
  PositionFactory.create() on every iteration.

### broker dispatch path
- Replace EntryBatchIndexesAcks.Pair<Integer, long[]>[] with two parallel
  arrays (int[] batchSizes, long[][] ackSets) so recording an entry's ack info
  no longer allocates a Pair.
- Add EntryBatchIndexesAcks.setIndexesAcks(int, int, long[]) as the new
  allocation-free API; keep the Pair-accepting overload as @deprecated for
  backwards compatibility with existing tests.
- AbstractBaseDispatcher: call the new overload.

## Testing
- :managed-ledger:compileJava / :compileTestJava
- :pulsar-broker:compileJava / :compileTestJava
- :pulsar-broker:test EntryBatchIndexesAcksTest
- :managed-ledger:test PositionRangeSetTest, ManagedCursorTest
All passed.
Increase LocalBookkeeperEnsemble from 1 to 3 bookies and replicas
in WorkerUtilsBookKeeperPackageTest. A single-bookie ensemble is
inherently fragile and intermittently fails with BKTransmitException
(error code -6, BookieHandleNotAvailableException) during ledger
close under load. This matches the pattern used by other similar
tests (e.g., PulsarFunctionAdminTest uses 3 bookies).

Signed-off-by: Asthenia0412 <contributor@pulsar.apache.org>
Removing NUM_BOOKIES constant and related changes that are unrelated
to the managed-ledger dispatch optimization fix.

Refs: inline review comment from lhotari on line 62.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants