Add serialized response coalescing for high fanout fetch workloads#31010
Add serialized response coalescing for high fanout fetch workloads#31010ballard26 wants to merge 7 commits into
Conversation
Allocate fetch memory units just before read_from_partition rather than at function entry, so they are no longer held longer than needed. Move strict_max_bytes down alongside it to keep the obligatory-read setup with the read.
There was a problem hiding this comment.
Pull request overview
This PR introduces a per-shard “fetch read coalescer” to reduce duplicate partition reads/serialization when many consumers fetch the same partition offset concurrently (high fanout). It wires the coalescer into the Kafka fetch path behind a new dynamic config flag, and adds both unit and end-to-end tests.
Changes:
- Add
fetch_read_coalescer(cache + in-flight sharing + metrics) and expose it viakafka::server. - Integrate coalescing into
do_read_from_ntpand adjust fetch response memory accounting to support shared results. - Add unit tests for the coalescer plus an rptest exercising coalescing via metrics.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| tools/type-checking/type-check-strictness.json | Adds the new rptest file to strict type checking inputs. |
| tests/rptest/tests/fetch_read_coalescing_test.py | New end-to-end fanout test validating coalescing engages and data correctness. |
| src/v/kafka/server/tests/fetch_test.cc | Updates test helper call to pass the new read_coalescer() dependency. |
| src/v/kafka/server/tests/fetch_read_coalescer_test.cc | New unit tests covering coalescer hit/miss, staleness, inflight sharing, errors, and enable/disable behavior. |
| src/v/kafka/server/tests/BUILD | Registers the new coalescer unit test target. |
| src/v/kafka/server/server.h | Adds the coalescer member and exposes read_coalescer() accessor. |
| src/v/kafka/server/server.cc | Constructs the coalescer with cache sizing and binds it to the new config property. |
| src/v/kafka/server/handlers/fetch.h | Extends read_result (record_count, shared memory units), and extends read_from_ntp signature to accept the coalescer. |
| src/v/kafka/server/handlers/fetch.cc | Implements coalesced read path (shared reads + cloning) and adapts memory-units handling. |
| src/v/kafka/server/fetch_read_coalescer.h | New coalescer public API/types (key, cache entry, metrics). |
| src/v/kafka/server/fetch_read_coalescer.cc | New coalescer implementation (cache/inflight sharing + metrics). |
| src/v/kafka/server/BUILD | Adds new coalescer sources/headers and dependency on chunked_kv_cache. |
| src/v/config/configuration.h | Adds kafka_fetch_read_coalescing_enabled configuration property declaration. |
| src/v/config/configuration.cc | Defines the new tunable property and its description/default. |
ballard26
force-pushed
the
fetch-read-coal
branch
from
a4136bc to
9f9c7cd
Compare
Change read_result::memory_units and the response placeholder's units slot from std::optional to std::shared_ptr, wrapping the units in make_shared on the read path. This lets a single set of units be shared by every response built from a coalesced read result, released when the last reference drops.
Adds a per-shard fetch read coalescer that dedups reads with the same (ktp, offset, isolation, max_bytes, obligatory) key. Obligatory reads are cached separately from non-obligatory reads as the in-flight read of non-obligatory reads can return empty results which isn't valid for obligatory ones. Although not tracked in the key the known end-offset of the partition is tracked along with the read_result. This lets the coalescer check if the currently cached result is "fresh" by checking whether the end-offset for that read is greater than or equal to the provided end-offset. If it's not fresh the cached read is replaced with a newer one.
Live (needs_restart::no) cluster flag, default off, gating the fetch read coalescer. Wired into the fetch path in a following commit.
Adds the per-shard fetch_read_coalescer to the server class. And then threads it through fetch_ntps into do_read_from_ntp. With the coalescing flag off (default) the fetch reads remain as they were before. When on, concurrent and back-to-back readers of the same (ntp, offset, isolation, max_bytes, obligatory) key share one serialized read. This shared read is kept alive until all responses containing it are fully sent.
Tests that a fanout workload with independent consumers reading the same partition offset with read coalescing enabled share their serialized response in the fetch path.
This commit is just to run the full CI against this config enabled. Will be dropped before merge.
ballard26
force-pushed
the
fetch-read-coal
branch
from
9f9c7cd to
5f1e1ab
Compare
ballard26
changed the title
StephanDollberg
left a comment
StephanDollberg
left a comment
There was a problem hiding this comment.
Can you summarize (and also include in the commit messages) why you in the end didn't go with the various approaches we discussed?
If I conclude correctly the advantage of this is now that it shares memory in serialized form so it avoids issues of the share-from-batchcache approach?
|
Having all these std::shared_ptrs everywhere really starts to get ugly. Do we even need them? Isn't the only thing that leaves the shard a shared-iobuf? |
| kafka::partition_proxy part, | ||
| ntp_fetch_config& ntp_config, | ||
| model::offset lso, | ||
| std::optional<model::timeout_clock::time_point> deadline, |
There was a problem hiding this comment.
How does the deadline interact with the cache? Can we have one fetch with a short deadline get aborted and that error propagating to other fetches that would have had a longer deadline?
| .max_bytes = ntp_config.cfg.max_bytes, | ||
| .obligatory = obligatory_batch_read}; | ||
|
|
||
| co_return get_result( |
There was a problem hiding this comment.
Is there a ticket / slack link with more details about why this design is chosen, what analysis was done and what alternatives were rejected? Do you have any CPU profiles / log line traces by any chance, showing that this specific section that we're caching is the expensive part in a high fanout scenario (and not, for example, the per-request overhead of parsing requests etc etc)? Or perhaps do you have some benchmarks showing the impact of this change already?
This PR introduces a per-shard fetch read coalescer to reduce duplicate partition reads/serialization when many consumers fetch the same partition offset concurrently (high fan-out). This feature is off by default, but can be dynamically enabled without restart via the
kafka_fetch_read_coalescing_enabledcluster config.Changes:
fetch_read_coalescerservice(cache + in-flight sharing + metrics).do_read_from_ntpand adjusts fetch response memory accounting to support shared results.Backports Required
Release Notes
Improvements