[improve][ml] Complete a read at the last confirmed entry instead of extending it on the ledger executor - #26743

Motivation

When a cursor read asks for more entries than the managed ledger has confirmed, ManagedLedgerImpl.internalReadFromLedger reads the current ledger only up to the last confirmed entry. OpReadEntry.checkReadCompletion then sees that more entries were confirmed while the read ran (cursor.hasMoreEntries()), and schedules another read on the managed ledger's executor to fill the requested count, repeating until the count is reached. Its comment says it is meant for continuing into the next ledger. These reads are served from the entry cache, so the whole read completion, including the dispatcher's work (Key_Shared's filterAndGroupEntriesForDispatching, sending to the consumers), then runs on the managed ledger's executor thread, which is also the thread that adds entries.

Profiling of the IoT telemetry max-rate performance scenario on master c5a001313e8e (500 producers, 128-byte unbatched messages, one topic, 20-member Key_Shared subscription) shows that thread, BookKeeperClientWorker-OrderedExecutor-12-0, as the serial stage: 91 % busy with dispatcherMaxReadBatchSize 100 and 97 % with 500 or 1000. Its CPU samples per million messages in these extended reads (under OpReadEntry.checkReadCompletion → ManagedLedgerImpl.asyncReadEntries) grow with the batch size, and account for most of the growth of the thread's cost per message:

This is why the throughput falls when dispatcherMaxReadBatchSize is raised beyond 500 (unprofiled means on master: 99.4k msg/s with 100, 106.6k with 250, 106.1k with 500, 102.4k with 750, 98.5k with 1000).

Modifications

• OpReadEntry.readUpToLastConfirmedEntry: whether the last ledger read of the operation ended at the managed ledger's last confirmed entry; ManagedLedgerImpl.internalReadFromLedger sets it for each ledger read, and when the read position is already past the last confirmed entry of the current ledger.
• OpReadEntry.checkReadCompletion completes such a read with the entries it already has. Reads that cross into the next ledger, and reads that have returned no entries yet (for example because all of them were filtered as individually deleted), continue as before.

The entries confirmed after the read are read by the dispatcher's next read, from the thread that issues it.

Measurements

tests/performance launcher on one host (Intel i9-9980HK, 8 cores, 16 hardware threads, turbo off, fixed 2.4 GHz, thermald stopped, cool-down to 55 °C before each run). Baseline master c5a001313e8e, candidate this change on top of it (9f888c13858e; this PR is the same change rebased onto a later master), each built into its own image; dispatcherMaxReadBatchSize set with --set cluster.brokers.env.dispatcherMaxReadBatchSize=<n>. The runs were interleaved in one series (master@500, change@500, change@1000, change@100, three rounds in rotating order), then the high-rate runs and two profiled runs. Every run delivered every message with no duplicates, ordering violations or invalid messages, and none throttled.

iot-telemetry-max-rate (4,000,000 measured 128-byte messages, 500 producers, 1 × 20 Key_Shared consumers, no rate limit), 3 unprofiled runs each:

Every run of this change at 500 or 1000 is faster than every master run at 500, and 1000 no longer falls behind 500 (on master in an earlier series: 106.1k msg/s at 500, 98.5k at 1000). With the default of 100 a read rarely asks for more entries than are confirmed, so the change has little to act on; there was no master@100 run in this series (99.4k msg/s in an earlier series, when master@500 measured 106.1k, so the host ran about 5 % slower in this series).

Charts of the median unprofiled max-rate run of each side at dispatcherMaxReadBatchSize 500, as the run report renders them (master c5a001313e8e: 100,109 msg/s; this change: 104,952 msg/s):

Latency by percentile, baseline (master, 500)

Latency by percentile, this change (500)

Throughput over time, baseline (master, 500)

Throughput over time, this change (500)

Backlog over time, baseline (master, 500)

Backlog over time, this change (500)

iot-telemetry-high-rate (30,000 msg/s, 5 × 10 Key_Shared consumers), 2 runs each at 500: publish p50 1.9 ms (master) and 1.8 ms (this change), end-to-end p50 8 ms on both, end-to-end p99 15–18 ms (master) and 17–19 ms (this change), within the run-to-run spread.

Profiled broker (max-rate, --extends configs/profile-broker), CPU samples per million messages (10 ms interval); master from 2 runs per size of an earlier series, this change from 1 run per size:

The extended reads and the dispatch they carried are gone from the managed-ledger thread. That thread stays the serial stage (about 95 % busy in the profiled runs of this change), now with the cost of the adds alone.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as ManagedCursorTest, NonDurableCursorTest, ManagedLedgerTest, ManagedLedgerErrorsTest and OpReadEntryConfigTest, which pass locally (436 tests). There is no new unit test for the tail-read case yet: it needs entries to be confirmed between the start and the completion of a cache read.

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: a read that ends at the last confirmed entry completes on the thread that performed it, instead of being extended on the managed ledger's executor, so the dispatcher's read completion no longer runs on the thread that adds entries in that case.

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