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

[lhotari]

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 once the dispatcher Flow contention is removed.

Modifications

Batch the adds across the thread boundary between the publishing threads and the ledger executor:

• asyncAddEntry appends the add to the ledger's add entry handover queue, 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 every add 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 runs at most managedLedgerAddEntryHandoverMaxBatchSize adds (default 1024) and schedules the next batch for the rest, so add completions and other executor tasks keep running under load. A failing add is logged and does not stop the batch. If the executor rejects the batch task, the flag is cleared and the caller fails as before; the add stays queued, but the managed ledger's executor rejects tasks only once it is shut down, after which no batch runs.
• 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 setting managedLedgerAddEntryHandoverMaxBatchSize (default 1024), passed to the managed ledger as ManagedLedgerConfig.addEntryHandoverMaxBatchSize. A managed ledger captures the value when it opens, so the add path reads no shared configuration; an update applies to managed ledgers opened after it. 0 disables batching, and each add is then submitted to the executor as a task of its own, as before this change. Negative values are rejected.

Trade-offs of the handover batch size

A batch runs to completion on the ledger's executor thread before any other task on that thread. A larger managedLedgerAddEntryHandoverMaxBatchSize reduces scheduling overhead and contention between publishing threads at high publish rates, but keeps 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. A smaller value favors that latency over add throughput. The setting is tunable so that this balance can be adapted to the workload.

Measurements

The Key_Shared 500×20 (iot-key-shared-500x20.yaml) and IoT telemetry 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 with ZGC. Baseline is the parent of this change (#26716, cd9726ba76f0, which applies Shared and Key_Shared Flow permits without the dispatcher monitor), candidate is this change (1e80d28e8956, with the default managedLedgerAddEntryHandoverMaxBatchSize of 1024). The measurements come from one rotation over the stack and its follow-up experiments, with the builds interleaved run by run: three unprofiled Key_Shared runs, two unprofiled high-rate runs and one profiled Key_Shared run (iot-key-shared-500x20-profile.yaml) 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 50–55 °C. None of the Key_Shared runs throttled thermally; in the high-rate scenario the host throttled thermally during both of the two runs of the baseline and both of the two runs of this change.

Setting	Key_Shared 500×20	IoT telemetry high rate
Cluster	1 broker, 3 bookies	1 broker, 3 bookies
Ledger replication	E=1, W=1, A=1	broker defaults
Producers	500 gateways × 1 topic	500 gateways × 1 topic
Applications	1 × 20 consumers, Key_Shared	5 × 10 consumers, Key_Shared
Messages	4,000,000 measured, 1,000,000 warmup	5,000,000 measured, 1,000,000 warmup
Payload	128 bytes, batching off	64 bytes, batching on
Rate limit	none	none

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

Key_Shared 500×20, unprofiled:

	Baseline (#26716)	This change	Change
Producer throughput (msg/s)	99,897 · 99,672 · 98,056 (mean 99,208)	103,273 · 105,937 · 106,469 (mean 105,226)	+6.1 %
Publish latency p50 (ms)	984.6 · 984.1 · 1,003.5	965.6 · 938.5 · 940.5	−4.3 %
End-to-end latency p50 (ms)	1,275.9 · 1,279.0 · 1,266.7	980.5 · 948.2 · 954.4	−24.6 %
End-to-end latency p99 (ms)	1,742.8 · 1,750.0 · 1,772.5	1,122.3 · 1,073.2 · 1,096.7	−37.5 %
Sampled maximum backlog	63,369 · 67,043 · 63,340	25,875 · 26,765 · 49,855	−47.1 %

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):

	Baseline (#26716)	This change	Change
Producer throughput (msg/s)	57,940 · 57,098 (mean 57,519)	61,371 · 61,680 (mean 61,526)	+7.0 %
Publish latency p50 (ms)	1,681.4 · 1,741.8	1,626.1 · 1,648.6	−4.3 %
End-to-end latency p50 (ms)	2,656–2,748 · 2,302–2,372	2,062–2,109 · 2,076–2,107	−17.4 %
End-to-end latency p99 (ms)	3,224–3,369 · 3,531–3,602	2,896–3,031 · 2,873–2,916	−14.7 %
Sampled maximum backlog per application	95,493–106,031 · 85,874–90,776	60,383–68,605 · 82,522–85,356	−21.0 %

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

	Baseline (#26716)	This change	Change
Producer throughput (msg/s)	93,363	101,116	+8.3 %
Blocked in GrowableBatchedArrayBlockingQueue.offer (off-CPU, observed)	0.55 s	0.003 s
Blocked time with an application frame (off-CPU, observed)	0.88 s	0.34 s
Ledger-thread CPU samples per million messages	1,015	913	−10.1 %
Broker CPU samples per million messages	6,913	6,389	−7.6 %

In the baseline, the connection threads blocking on the ledger executor's queue lock from ServerCnx.handleSend are 62 % of the broker's blocked time with an application frame; with this change that wait is practically 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. End-to-end latency now follows the publish latency closely (the Key_Shared end-to-end p50 is on average 13 ms above the publish p50, against 283 ms in the baseline), most likely because dispatch work on the ledger executor no longer queues behind one task per published message, and the subscription's backlog stays smaller. Both scenarios improve: the Key_Shared throughput by 6.1 % and the high-rate throughput by 7.0 %, with every run of this change faster than every baseline run in both.

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 Key_Shared run of each side (baseline 99,672 msg/s, this change 105,937 msg/s), as the run report renders them:

Baseline (#26716)	This change

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• ManagedLedgerTest.testConcurrentAsyncAddEntriesKeepPerThreadOrder: 8 threads each add 2,000 entries concurrently to a fresh ledger, racing to create the queue; all adds complete and each thread's entries are written in the order that thread added them.
• ManagedLedgerTest.testAddEntryHandoverQueueIsCreatedByTheFirstAdd: an opened ledger has no add entry handover queue until the first add.
• ManagedLedgerTest.testConcurrentAsyncAddEntriesKeepPerThreadOrder runs with batch sizes 0, 1 and 1024; testAddEntryHandoverBatchingDisabled, testAddEntryHandoverMaxBatchSizeIsCapturedWhenOpened and testAddEntryHandoverMaxBatchSizeRejectsNegativeValues cover the disabled mode, the value captured at open and validation.
• BrokerServiceTest.testManagedLedgerAddEntryHandoverMaxBatchSizeConfiguration and testManagedLedgerAddEntryHandoverMaxBatchSizeDynamicUpdate: the broker setting reaches the managed ledger configuration, and a dynamic update is 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, 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.