[improve][ml] Batch managed-ledger adds across the thread boundary to the ledger executor - #26717

Motivation

ManagedLedgerImpl.asyncAddEntry submits one executor task per entry to the managed ledger's ordered executor. With hundreds of producer connections on one topic, the connection threads contend on the executor's queue lock (BookKeeper's GrowableBatchedArrayBlockingQueue.offer → ReentrantLock.lock) once per message. The ledger thread also runs one task per message, and other work on that thread, such as waiting-cursor wake-ups for tailing dispatchers, queues behind those tasks.

Off-CPU profiling of 500 producers and a 20-member Key_Shared subscription shows this queue lock as the largest blocked time with an application frame in the broker after the dispatcher Flow contention, which #26716 removes.

Modifications

Batch the adds across the thread boundary between the publishing threads and the ledger executor, through a new package-private BatchingExecutorWrapper in the managed ledger module:

• asyncAddEntry hands the add to the wrapper, which appends it to a JCTools MpscUnboundedArrayQueue (already used by RangeCacheRemovalQueue). The thread that finds no batch task scheduled submits one to the ledger executor, and that task runs the adds queued by then. Concurrent publishers contend on the executor's queue once per batch instead of once per add, and other executor work no longer waits behind one task per published message.
• A batch stops after managedLedgerAddEntryHandoverMaxBatchItems adds (default 1024), or once the entries it has run add up to managedLedgerAddEntryHandoverMaxBatchBytesSize bytes (default 5 MB), and schedules the next batch for the rest, so add completions and other executor tasks keep running under load. The byte limit keeps a managed ledger with large entries from holding the executor thread, which is shared with other managed ledgers, for as long as a full batch of large entries would. A batch always runs at least one add, even one larger than the byte limit. The wrapper is generic: a task implementing BatchingExecutorWrapper.WeightedRunnable weighs getWeight(), and the managed ledger's add task weighs the readable bytes of its entry. That task replaces the lambda that captured the same fields, so it allocates nothing more. A failing add is logged and does not stop the batch.
• The thread that sets the batch flag is the only consumer of the queue until it clears it, including while its batch runs, and it clears the flag before checking the queue again, so no add is left behind. If the executor rejects a batch task, that thread takes the queued adds: the caller's own add does not run and the caller fails with the executor's exception, as before this change, and every other queued add is failed through its callback with its buffer released. asyncAddEntry also releases the buffer it retained for an add that the executor rejects.
• The queue is created by the first add, with a compare-and-set on a volatile field, so managed ledgers that are never written to do not allocate it; it is never replaced, so racing first adds all use the same queue. It uses 512-entry chunks and links another chunk only when a batch backs up beyond that: about 2.7 KB per managed ledger that has been written to.
• Adds are still created and processed on the ledger executor, so the ledger's threading and each thread's add order are unchanged.
• New dynamic broker settings managedLedgerAddEntryHandoverMaxBatchItems (default 1024) and managedLedgerAddEntryHandoverMaxBatchBytesSize (default 5242880), passed to the managed ledger as ManagedLedgerConfig.addEntryHandoverMaxBatchItems and addEntryHandoverMaxBatchBytesSize. A managed ledger captures the values when it opens, so the add path reads no shared configuration; an update applies to managed ledgers opened after it. A max batch items of 0 or 1 disables batching, and each add is then submitted to the executor as a task of its own, as before this change. A max batch bytes size of 0 removes the byte limit. Negative values are rejected.

Trade-offs of the handover batch limits

A batch runs to completion on the ledger's executor thread before any other task on that thread. Larger limits reduce scheduling overhead and contention between publishing threads at high publish rates, but keep the executor thread occupied for longer per batch, which can delay add completions, reads and cursor notifications for the managed ledgers that share the thread. Smaller limits favor that latency over add throughput. The item limit bounds the per-add overhead of a batch, and the byte limit bounds the work of a batch of large entries, so that on a heavily loaded broker or bookies a managed ledger with very large entries does not starve managed ledgers with small entries that share its executor thread. Both are tunable so that this balance can be adapted to the workload.

Measurements

The IoT telemetry max-rate (iot-telemetry-max-rate.yaml, 500 producers and a 20-member Key_Shared subscription, formerly iot-key-shared-500x20.yaml) and high-rate (iot-telemetry-high-rate.yaml) performance scenarios of tests/performance, run with the performance launcher on one host (Intel i9-9980HK, 8 cores, 16 hardware threads) running the broker, 3 bookies and the clients, on JDK 25.0.4.1 with ZGC. The host was configured with the performance testing environment setup of tests/performance/environment, whose TuneD profile disables turbo, so that the CPU ran at a fixed 2.4 GHz in every run. Baseline is the parent of this change, master at f925a25884e6 (#26715), and the candidate is the head of this change, 87c32c1213c9, with the default limits of 1024 adds and 5 MB per handover batch. Both were built from clean checkouts into separate images. The builds were interleaved run by run: three unprofiled max-rate runs (baseline and candidate, then candidate and baseline, then baseline and candidate), two unprofiled high-rate runs (baseline first, then candidate first) and one profiled max-rate run with --extends configs/profile-broker of each build. thermald was stopped, and the launcher let the CPU package cool down to 55 °C before each run and again before each measurement (-Pperformance.cooldownTemperature=55), so every run started at 48–53 °C. None of the runs throttled thermally. The metrics stack of tests/performance scraped the broker, the bookies and ZooKeeper every 5 s during each run, with the broker's stats periods set to the same 5 s.

Every run delivered all messages to every application with no duplicates, no ordering violations and no invalid messages.

IoT telemetry max rate, unprofiled (runs in the order they ran):

IoT telemetry high rate, unprofiled (end-to-end latency and backlog are the ranges over the 5 applications, and their change is that of the mean over the applications):

Broker, profiled max-rate run (JFR recording, flame graphs and off-CPU profile of the measurement period, 4,000,000 messages on each side):

In the baseline, the connection threads blocking on the ledger executor's queue lock from ServerCnx.handleSend are 30.8 % of the broker's blocked time with an application frame, the largest single wait, next to the dispatcher monitor waits in internalConsumerFlow (56 % over its two rows) that #26716 removes; with this change that wait is gone.

The managed-ledger thread (BookKeeperClientWorker-OrderedExecutor-12-0) stays the serial stage, so throughput follows its cost per message, which drops with fewer executor tasks. The max-rate throughput improves by 11.0 %, with every run of this change faster than every baseline run. End-to-end latency now follows the publish latency closely (the max-rate end-to-end p50 is on average 13 ms above the publish p50, against 290 ms in the baseline), most likely because dispatch work on the ledger executor no longer queues behind one task per published message.

The high-rate scenario publishes at a fixed 30,000 messages per second, which this host sustains on both sides, so it shows the latency at a rate below saturation rather than the maximum throughput. Both sides deliver every message at that rate with a publish p50 of 2.1–2.2 ms and an end-to-end p50 of 8 ms. The end-to-end p99 of 19–25 ms and the backlogs of at most 5,300 messages vary between the runs of each side by more than the difference between the sides: the first run of this change had the highest backlogs, with a p99 within the baseline's range, and its second run the highest p99, with backlogs within the baseline's range.

In an earlier session, an eagerly created queue with 256-entry chunks and the final lazily created queue with 512-entry chunks were compared in 3 interleaved unprofiled runs each (means 110.7k and 108.7k msg/s, within the run-to-run spread).

Charts of the median unprofiled max-rate run of each side (baseline 90,509 msg/s, this change 99,923 msg/s), as the run report renders them:

Baseline (#26715)	This change

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• BatchingExecutorWrapperTest (unit tests with a hand-driven delegate):
  • adds queued before a batch runs share one batch;
  • a batch stops at the item limit, and at the weight limit, and the rest runs in the next batch;
  • a task heavier than the weight limit runs in a batch of its own, and unweighted tasks are limited only by the item limit;
  • a task queued while a batch runs is not left behind, and a failing task does not stop the batch;
  • the queue is created by the first task;
  • a rejected caller fails and its task does not run later; tasks queued meanwhile by other threads, and those left for a rejected follow-up batch, go to the rejected task handler;
  • invalid limits are rejected.
• BatchingExecutorWrapperTest stress tests: 8 threads submitting 10,000 tasks each keep their per-thread order (with item limits 2 and 1024, and with a weight limit); against a delegate that always rejects, every task is rejected and none runs; against a delegate that rejects every third batch, every task runs or is rejected exactly once.
• ManagedLedgerTest:
  • testAddEntryHandoverLimitsAreCapturedWhenOpened, testAddEntryHandoverWithoutByteLimit and testAddEntryHandoverBatchingDisabledUsesTheExecutor: a ledger passes its limits to the wrapper, a byte limit of 0 removes the weight limit, max batch items of 0 or 1 hands adds straight to the executor, and setConfig does not change the limits of an open ledger.
  • testAddEntryWithAddEntryHandoverBatchingDisabled and testAsyncAddEntriesWithSmallAddEntryHandoverMaxBatchBytesSize: adds complete in order with batching disabled, and with a byte limit that gives every add a batch of its own.
  • testRejectedAddEntryReleasesTheRetainedBuffer: with batching enabled and disabled, an add rejected by a shut down executor fails its caller and releases the buffer it retained.
  • testAddEntryHandoverMaxBatchLimitsRejectNegativeValues: defaults and validation of the configuration.
• BrokerServiceTest.testManagedLedgerAddEntryHandoverMaxBatchItemsConfiguration, testManagedLedgerAddEntryHandoverMaxBatchBytesSizeConfiguration and testManagedLedgerAddEntryHandoverMaxBatchItemsDynamicUpdate: the broker settings reach the managed ledger configuration, and dynamic updates are validated and applied.
• ManagedLedgerErrorsTest (22 cases) passes locally; it exercises add failures and recovery through the changed hand-off.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• 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

Threading model: adds reach the managed ledger's executor in batches bounded by a number of adds and a total entry size, through an MPSC queue run by one batch task at a time, instead of one executor task per add. The adds still run on the same executor thread, in the same per-thread order.

This PR was prepared with AI assistance (Claude Code) and reviewed by a human contributor.