From a9d062fa15bd0a22f1e6dff294d8fe24677db948 Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Thu, 2 Jul 2026 15:56:23 -0400 Subject: [PATCH 1/7] kafka: move memory unit allocation below the read preamble 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. --- src/v/kafka/server/handlers/fetch.cc | 42 ++++++++++++++-------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/src/v/kafka/server/handlers/fetch.cc b/src/v/kafka/server/handlers/fetch.cc index 7a5a7827ab21f..d3e5dbe912f88 100644 --- a/src/v/kafka/server/handlers/fetch.cc +++ b/src/v/kafka/server/handlers/fetch.cc @@ -217,27 +217,6 @@ static ss::future do_read_from_ntp( std::optional deadline, const bool obligatory_batch_read, fetch_memory_units_manager& units_mgr) { - // If it's the obligatory batch read then we need to allow for the - // configured max bytes to exceeded if the next batch in the partition - // is larger. This is needed to conform with KIP-74. - ntp_config.cfg.strict_max_bytes = !obligatory_batch_read; - - // control available memory - auto memory_units = units_mgr.zero_units(); - if (!ntp_config.cfg.skip_read) { - memory_units = units_mgr.allocate_memory_units( - ntp_config.ktp(), - ntp_config.cfg.max_bytes, - ntp_config.cfg.max_batch_size, - ntp_config.cfg.avg_batch_size, - obligatory_batch_read); - if (!memory_units.has_units()) { - ntp_config.cfg.skip_read = true; - } else if (ntp_config.cfg.max_bytes > memory_units.num_units()) { - ntp_config.cfg.max_bytes = memory_units.num_units(); - } - } - /* * lookup the ntp's partition */ @@ -314,6 +293,27 @@ static ss::future do_read_from_ntp( preferred_replica); } } + // If it's the obligatory batch read then we need to allow for the + // configured max bytes to exceeded if the next batch in the partition + // is larger. This is needed to conform with KIP-74. + ntp_config.cfg.strict_max_bytes = !obligatory_batch_read; + + // control available memory + auto memory_units = units_mgr.zero_units(); + if (!ntp_config.cfg.skip_read) { + memory_units = units_mgr.allocate_memory_units( + ntp_config.ktp(), + ntp_config.cfg.max_bytes, + ntp_config.cfg.max_batch_size, + ntp_config.cfg.avg_batch_size, + obligatory_batch_read); + if (!memory_units.has_units()) { + ntp_config.cfg.skip_read = true; + } else if (ntp_config.cfg.max_bytes > memory_units.num_units()) { + ntp_config.cfg.max_bytes = memory_units.num_units(); + } + } + auto result = co_await read_from_partition( std::move(*kafka_partition), maybe_lso.value(), ntp_config.cfg, deadline); From a35deeb2da4a89145917c888298f5868e8012589 Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Thu, 2 Jul 2026 16:12:12 -0400 Subject: [PATCH 2/7] kafka: hold fetch memory units via shared_ptr in read_result 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. --- src/v/kafka/server/handlers/fetch.cc | 11 ++++++----- src/v/kafka/server/handlers/fetch.h | 15 ++++++++------- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/src/v/kafka/server/handlers/fetch.cc b/src/v/kafka/server/handlers/fetch.cc index d3e5dbe912f88..5d0756d19bbe6 100644 --- a/src/v/kafka/server/handlers/fetch.cc +++ b/src/v/kafka/server/handlers/fetch.cc @@ -321,7 +321,8 @@ static ss::future do_read_from_ntp( // happen because there is no strict limit on read size when reading the // obligatory batch. memory_units.adjust_units(result.data_size_bytes()); - result.memory_units = std::move(memory_units); + result.memory_units = std::make_shared( + std::move(memory_units)); co_return result; } @@ -414,7 +415,7 @@ static void fill_fetch_responses( resp.preferred_read_replica = *res.preferred_replica; } - std::optional resp_units{}; + std::shared_ptr resp_units; auto current_response_size = resp_it->response_size(); auto bytes_left = octx.bytes_left - current_response_size; @@ -1750,10 +1751,10 @@ ss::future op_context::send_error_response(error_code ec) && { } ss::deleter op_context::response_memory_units_deleter() { - chunked_vector mu; + chunked_vector> mu; for (auto& r : iteration_order) { if (r.has_memory_units()) { - mu.push_back(r.release_memory_units().value()); + mu.push_back(r.release_memory_units()); } } return ss::make_object_deleter(std::move(mu)); @@ -1775,7 +1776,7 @@ op_context::response_placeholder::response_placeholder( void op_context::response_placeholder::set( fetch_response::partition_response&& response, - std::optional&& response_memory_units) { + std::shared_ptr&& response_memory_units) { vassert( response.partition_index == _it->partition_response->partition_index, "Response and current partition ids have to be the same. Current " diff --git a/src/v/kafka/server/handlers/fetch.h b/src/v/kafka/server/handlers/fetch.h index 2a92dd1f64f37..a5f9862b81dde 100644 --- a/src/v/kafka/server/handlers/fetch.h +++ b/src/v/kafka/server/handlers/fetch.h @@ -52,7 +52,7 @@ struct op_context { void set( fetch_response::partition_response&&, - std::optional&&); + std::shared_ptr&&); const model::topic& topic() { return _it->partition->topic; } model::partition_id partition_id() { @@ -90,13 +90,13 @@ struct op_context { } // Adds/replaces the memory units that are held for this ntp. - void - replace_or_add_memory_units(std::optional&& units) { + void replace_or_add_memory_units( + std::shared_ptr&& units) { _response_memory_units = std::move(units); } // Releases/returns the memory units that are held for this ntp. - std::optional release_memory_units() { + std::shared_ptr release_memory_units() { return std::move(_response_memory_units); } @@ -104,8 +104,9 @@ struct op_context { fetch_response::iterator _it; op_context* _ctx; const model::ktp_with_hash _ktp; - // Tracks memory used by response data in `_it`. - std::optional _response_memory_units; + // Tracks memory used by response data in `_it`. Shared so a coalesced + // read result can hand the same units to every response sharing it. + std::shared_ptr _response_memory_units; }; using iteration_order_t @@ -367,7 +368,7 @@ struct read_result { error_code error; model::partition_id partition; std::vector aborted_transactions; - std::optional memory_units; + std::shared_ptr memory_units; }; // struct aggregating fetch requests and corresponding response iterators for // the same shard From 6a719884e42f7e749002c0a481213a290cabe1f6 Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Thu, 2 Jul 2026 19:38:10 -0400 Subject: [PATCH 3/7] kafka: add fetch_read_coalescer 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. --- src/v/kafka/server/BUILD | 4 + src/v/kafka/server/fetch_read_coalescer.cc | 156 +++++++++ src/v/kafka/server/fetch_read_coalescer.h | 141 ++++++++ src/v/kafka/server/tests/BUILD | 18 + .../server/tests/fetch_read_coalescer_test.cc | 321 ++++++++++++++++++ 5 files changed, 640 insertions(+) create mode 100644 src/v/kafka/server/fetch_read_coalescer.cc create mode 100644 src/v/kafka/server/fetch_read_coalescer.h create mode 100644 src/v/kafka/server/tests/fetch_read_coalescer_test.cc diff --git a/src/v/kafka/server/BUILD b/src/v/kafka/server/BUILD index 32779a58c1260..0807915d1c54f 100644 --- a/src/v/kafka/server/BUILD +++ b/src/v/kafka/server/BUILD @@ -123,6 +123,7 @@ redpanda_cc_library( "data_migration_group_proxy_impl.cc", "fetch_memory_units.cc", "fetch_pid_controller.cc", + "fetch_read_coalescer.cc", "fetch_session_cache.cc", "group.cc", "group_initializer.cc", @@ -180,6 +181,7 @@ redpanda_cc_library( "fetch_memory_units.h", "fetch_metadata_cache.h", "fetch_pid_controller.h", + "fetch_read_coalescer.h", "fetch_session.h", "fetch_session_cache.h", "fwd.h", @@ -328,6 +330,7 @@ redpanda_cc_library( "//src/v/container:intrusive", "//src/v/datalake:partition_spec_parser", "//src/v/features", + "//src/v/hashing:combine", "//src/v/hashing:jump_consistent", "//src/v/hashing:xx", "//src/v/kafka/data:partition_proxy", @@ -457,6 +460,7 @@ redpanda_cc_library( "//src/v/strings:string_switch", "//src/v/strings:utf8", "//src/v/utils:absl_sstring_hash", + "//src/v/utils:chunked_kv_cache", "//src/v/utils:ema", "//src/v/utils:log_hist", "//src/v/utils:moving_average", diff --git a/src/v/kafka/server/fetch_read_coalescer.cc b/src/v/kafka/server/fetch_read_coalescer.cc new file mode 100644 index 0000000000000..40279a7262e2e --- /dev/null +++ b/src/v/kafka/server/fetch_read_coalescer.cc @@ -0,0 +1,156 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Use of this software is governed by the Business Source License + * included in the file licenses/BSL.md + * + * As of the Change Date specified in that file, in accordance with + * the Business Source License, use of this software will be governed + * by the Apache License, Version 2.0 + */ +#include "kafka/server/fetch_read_coalescer.h" + +#include "base/vassert.h" +#include "config/configuration.h" +#include "hashing/combine.h" +#include "kafka/server/handlers/fetch.h" +#include "metrics/prometheus_sanitize.h" + +namespace kafka { + +namespace { + +// Serve a retained result only while the visible end has not advanced past the +// point it was read at; otherwise a re-read must pick up the new data. +bool is_fresh( + const read_result& r, + model::isolation_level level, + model::offset current_bound) { + auto bound = level == model::isolation_level::read_committed + ? r.last_stable_offset + : r.high_watermark; + return current_bound <= bound; +} + +} // namespace + +size_t coalesce_key_hash::operator()(const coalesce_key& k) const { + size_t h = std::hash{}(k.ktp); + hash::combine( + h, + k.fetch_offset(), + static_cast(k.level), + k.max_bytes, + k.obligatory); + return h; +} + +fetch_read_coalescer::fetch_read_coalescer( + cache_config cfg, config::binding enabled) + : _cache_config(cfg) + , _enabled(std::move(enabled)) { + if (_enabled()) { + _cache.emplace(_cache_config); + } + _enabled.watch([this] { on_enabled_change(); }); + setup_metrics(); +} + +void fetch_read_coalescer::on_enabled_change() { + if (_enabled() && !_cache) { + _cache.emplace(_cache_config); + } else if (!_enabled() && _cache) { + _cache.reset(); + } +} + +ss::future fetch_read_coalescer::get_or_insert( + const coalesce_key& key, model::offset current_bound, read_fn fn) { + vassert(_cache.has_value(), "coalescer used while disabled"); + ss::shared_ptr entry; + if (auto existing = _cache->get_value(key)) { + entry = *existing; + if (entry->inflight) { + ++_inflight_hits; + return entry->inflight->get_shared_future(); + } + if ( + auto r = entry->ready.lock(); + r && is_fresh(*r, key.level, current_bound)) { + ++_ready_hits; + return ss::make_ready_future(std::move(r)); + } + ++_reinsertions; + } else { + entry = ss::make_shared(); + _cache->try_insert(key, entry); + ++_insertions; + } + return read_and_publish(std::move(entry), std::move(fn)); +} + +void fetch_read_coalescer::coalesce_entry::begin_read() { + inflight.emplace(); + ready.reset(); +} + +void fetch_read_coalescer::coalesce_entry::finish_read(shared_read result) { + // Retain only data-bearing reads. An empty read drains instantly, so a + // retained empty would expire before reuse; concurrent empty reads still + // coalesce via the in-flight promise. + if (result->error == error_code::none && result->has_data()) { + ready = result; + } + inflight->set_value(std::move(result)); + inflight.reset(); +} + +void fetch_read_coalescer::coalesce_entry::fail_read(std::exception_ptr e) { + inflight->set_exception(std::move(e)); + inflight.reset(); +} + +ss::future fetch_read_coalescer::read_and_publish( + ss::shared_ptr entry, read_fn fn) { + entry->begin_read(); + try { + auto shared = std::make_shared(co_await fn()); + entry->finish_read(shared); + co_return std::move(shared); + } catch (...) { + entry->fail_read(std::current_exception()); + throw; + } +} + +void fetch_read_coalescer::setup_metrics() { + if (config::shard_local_cfg().disable_metrics()) { + return; + } + namespace sm = ss::metrics; + _metrics.add_group( + prometheus_sanitize::metrics_name("kafka:fetch_read_coalescer"), + { + sm::make_counter( + "insertions_total", + [this] { return _insertions; }, + sm::description("Fetch reads that inserted a new cache entry.")), + sm::make_counter( + "reinsertions_total", + [this] { return _reinsertions; }, + sm::description( + "Fetch reads that re-read an existing (stale or expired) entry.")), + sm::make_counter( + "ready_hits_total", + [this] { return _ready_hits; }, + sm::description("Fetch reads served from a retained result.")), + sm::make_counter( + "inflight_hits_total", + [this] { return _inflight_hits; }, + sm::description("Fetch reads served by awaiting an in-flight read.")), + }, + {}, + {sm::shard_label}); +} + +} // namespace kafka diff --git a/src/v/kafka/server/fetch_read_coalescer.h b/src/v/kafka/server/fetch_read_coalescer.h new file mode 100644 index 0000000000000..f6ff3f2851f62 --- /dev/null +++ b/src/v/kafka/server/fetch_read_coalescer.h @@ -0,0 +1,141 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Use of this software is governed by the Business Source License + * included in the file licenses/BSL.md + * + * As of the Change Date specified in that file, in accordance with + * the Business Source License, use of this software will be governed + * by the Apache License, Version 2.0 + */ +#pragma once + +#include "base/seastarx.h" +#include "config/property.h" +#include "metrics/metrics.h" +#include "model/fundamental.h" +#include "model/ktp.h" +#include "model/metadata.h" +#include "utils/chunked_kv_cache.h" + +#include +#include +#include + +#include +#include + +namespace kafka { + +// Defined in kafka/server/handlers/fetch.h. +struct read_result; + +using shared_read = std::shared_ptr; + +/// Same-key reads have byte-identical output. max_bytes is the requested +/// budget, sampled before allocation. +struct coalesce_key { + // Build from the config's ktp_with_hash, not the sliced ktp(), to reuse + // the planning-time hash. + model::ktp_with_hash ktp; + model::offset fetch_offset; + model::isolation_level level; + size_t max_bytes; + // Obligatory (non-strict) reads guarantee >=1 batch; strict reads honor + // max_bytes. Keyed apart so neither cross-serves the other, at the cost + // of a duplicate read when one (ntp, offset) is fetched both ways at once. + bool obligatory; + + friend bool operator==(const coalesce_key& a, const coalesce_key& b) { + return a.fetch_offset == b.fetch_offset && a.level == b.level + && a.max_bytes == b.max_bytes && a.obligatory == b.obligatory + && a.ktp == b.ktp; + } +}; + +struct coalesce_key_hash { + // Reuses ktp's cached hash; mixes in the other fields. + size_t operator()(const coalesce_key&) const; +}; + +/// Per-shard fetch read coalescer. Reads and serializes each unique key once, +/// then shares the result with concurrent and back-to-back readers of that key. +class fetch_read_coalescer { + /// Coalescing state for one key: an in-flight read that concurrent readers + /// await, plus a weakly retained completed read that later readers reuse. + struct coalesce_entry { + // Set while a read for this key is in flight; concurrent readers await + // its future instead of starting their own. Reset once the read + // resolves. + std::optional> inflight; + // The last completed read, held weakly so the coalescer never pins it. + std::weak_ptr ready; + + // A read is starting: install the promise waiters block on, and drop + // any retained result since the fresh read supersedes it. + void begin_read(); + // The read produced `result`: hand it to the waiters and retain it for + // later readers. + void finish_read(shared_read result); + // The read failed: hand the exception to the waiters and retain + // nothing. + void fail_read(std::exception_ptr); + }; + +public: + using cache_t = utils:: + chunked_kv_cache; + using cache_config = cache_t::config; + + /// Always constructed; disabling clears the cache via the binding watch. + fetch_read_coalescer(cache_config, config::binding enabled); + fetch_read_coalescer(const fetch_read_coalescer&) = delete; + fetch_read_coalescer& operator=(const fetch_read_coalescer&) = delete; + fetch_read_coalescer(fetch_read_coalescer&&) = delete; + fetch_read_coalescer& operator=(fetch_read_coalescer&&) = delete; + ~fetch_read_coalescer() = default; + + bool enabled() const { return _enabled(); } + + using read_fn = ss::noncopyable_function()>; + + /// Checks if there is a cached shared read for the provided `key`. + /// One of the following actions is taken internally: + /// - If there is a ready read for the key and that read's end offset is + /// greater than or equal to `current_bound` then that read is returned. + /// - If there is an in-flight read for the key it is awaited and the + /// resulting read is returned. + /// - Otherwise read_fn runs inline and its result is shared and retained. + /// Note that read_fn runs at most once, only on the miss path. If it throws + /// the exception is propagated to all waiters. + /// + /// The returned shared_read is the only strong owner of the result. The + /// cache keeps a weak handle so it never pins memory, so the caller must + /// hold the returned ref for as long as the result is needed (in the fetch + /// path this is until the response is sent to the client). + ss::future + get_or_insert(const coalesce_key&, model::offset current_bound, read_fn); + +private: + // Run read_fn inline on `entry` and share its result with the entry's + // waiters, retaining it iff it is data-bearing and error-free. + ss::future + read_and_publish(ss::shared_ptr, read_fn); + + void on_enabled_change(); + void setup_metrics(); + + cache_config _cache_config; + config::binding _enabled; + // Held via optional because chunked_kv_cache has no clear() and deletes its + // move; disable reset()s it. + std::optional _cache; + + uint64_t _insertions{0}; + uint64_t _reinsertions{0}; + uint64_t _ready_hits{0}; + uint64_t _inflight_hits{0}; + metrics::internal_metric_groups _metrics; +}; + +} // namespace kafka diff --git a/src/v/kafka/server/tests/BUILD b/src/v/kafka/server/tests/BUILD index 6422b947b420f..f4ec2e7feb399 100644 --- a/src/v/kafka/server/tests/BUILD +++ b/src/v/kafka/server/tests/BUILD @@ -244,6 +244,24 @@ redpanda_cc_gtest( ], ) +redpanda_cc_gtest( + name = "fetch_read_coalescer_test", + timeout = "short", + srcs = [ + "fetch_read_coalescer_test.cc", + ], + cpu = 2, + deps = [ + "//src/v/bytes:iobuf", + "//src/v/config", + "//src/v/kafka/server", + "//src/v/model", + "//src/v/test_utils:gtest", + "@googletest//:gtest", + "@seastar", + ], +) + redpanda_cc_gtest( name = "consumer_group_recovery_test", timeout = "short", diff --git a/src/v/kafka/server/tests/fetch_read_coalescer_test.cc b/src/v/kafka/server/tests/fetch_read_coalescer_test.cc new file mode 100644 index 0000000000000..b39efeac6b2ad --- /dev/null +++ b/src/v/kafka/server/tests/fetch_read_coalescer_test.cc @@ -0,0 +1,321 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Use of this software is governed by the Business Source License + * included in the file licenses/BSL.md + * + * As of the Change Date specified in that file, in accordance with + * the Business Source License, use of this software will be governed + * by the Apache License, Version 2.0 + */ +#include "bytes/iobuf.h" +#include "config/mock_property.h" +#include "kafka/server/fetch_read_coalescer.h" +#include "kafka/server/handlers/fetch.h" +#include "model/fundamental.h" +#include "model/ktp.h" +#include "model/metadata.h" +#include "test_utils/test.h" + +#include +#include + +namespace kafka { + +namespace { + +fetch_read_coalescer::cache_config test_cache_config() { + return {.cache_size = 8, .small_size = 1}; +} + +coalesce_key make_key( + model::offset offset = model::offset{5}, + size_t max_bytes = 1024, + model::isolation_level level = model::isolation_level::read_uncommitted, + bool obligatory = true) { + return { + model::ktp_with_hash{"topic", 0}, offset, level, max_bytes, obligatory}; +} + +// A data-bearing result with `bytes` of payload, read at the given bounds. +read_result make_result( + size_t bytes = 7, + model::offset hwm = model::offset{100}, + model::offset lso = model::offset{100}) { + iobuf data; + data.append(ss::temporary_buffer(bytes)); + return read_result( + std::make_unique(std::move(data)), + model::offset{5}, + model::offset{5}, + model::offset{5}, + 1, + hwm, + lso, + std::nullopt, + {}); +} + +// get_or_insert is the only public entry point, so tests drive it and prove +// coalescing behaviorally. Each read_fn bumps `calls`, so a shared serve leaves +// the count untouched and a miss increments it. read_fn is single-use +// (move-only), so a fresh one is built per get_or_insert call. +fetch_read_coalescer::read_fn counting_read( + int& calls, + size_t bytes = 7, + model::offset hwm = model::offset{100}, + model::offset lso = model::offset{100}) { + return [&calls, bytes, hwm, lso] { + ++calls; + return ss::make_ready_future(make_result(bytes, hwm, lso)); + }; +} + +fetch_read_coalescer::read_fn counting_error(int& calls) { + return [&calls] { + ++calls; + return ss::make_ready_future(read_result( + error_code::offset_out_of_range, + model::offset{5}, + model::offset{100}, + model::offset{100})); + }; +} + +fetch_read_coalescer::read_fn counting_empty(int& calls) { + return [&calls] { + ++calls; + return ss::make_ready_future(read_result( + model::offset{5}, model::offset{100}, model::offset{100})); + }; +} + +} // namespace + +TEST_CORO(FetchReadCoalescer, RetainAndShare) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + auto sr = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 1); + + // A later reader at the same (fresh) bound shares the retained result + // verbatim without re-reading — while a strong ref (sr) still holds it. + auto shared = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 1); + EXPECT_EQ(sr.get(), shared.get()); + + // Dropping the last strong ref expires the weak cache entry, so the next + // reader misses and re-reads. + sr = nullptr; + shared = nullptr; + co_await c.get_or_insert(key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 2); +} + +TEST_CORO(FetchReadCoalescer, DistinctKeysDoNotCoalesce) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + + int calls = 0; + // Hold every result so retention can't lapse; distinct keys must still each + // drive their own read. + std::vector held; + held.push_back( + co_await c.get_or_insert( + make_key(), model::offset{100}, counting_read(calls))); + EXPECT_EQ(calls, 1); + + // Different offset, max_bytes, isolation level, or obligatory flag -> a + // different key -> a fresh read, never a share. + held.push_back( + co_await c.get_or_insert( + make_key(model::offset{6}), model::offset{100}, counting_read(calls))); + EXPECT_EQ(calls, 2); + held.push_back( + co_await c.get_or_insert( + make_key(model::offset{5}, 2048), + model::offset{100}, + counting_read(calls))); + EXPECT_EQ(calls, 3); + held.push_back( + co_await c.get_or_insert( + make_key( + model::offset{5}, 1024, model::isolation_level::read_committed), + model::offset{100}, + counting_read(calls))); + EXPECT_EQ(calls, 4); + held.push_back( + co_await c.get_or_insert( + make_key( + model::offset{5}, + 1024, + model::isolation_level::read_uncommitted, + false), + model::offset{100}, + counting_read(calls))); + EXPECT_EQ(calls, 5); +} + +TEST_CORO(FetchReadCoalescer, InflightHitSharesOneRead) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + ss::promise block; + // Initiator: its read_fn runs but suspends on `block`, leaving the read in + // flight. + auto initiator = c.get_or_insert(key, model::offset{100}, [&calls, &block] { + ++calls; + return block.get_future(); + }); + + // A concurrent reader awaits the in-flight read rather than starting a + // second one, so its read_fn never runs. + auto waiter = c.get_or_insert( + key, model::offset{100}, counting_read(calls)); + + block.set_value(make_result()); + auto sr_initiator = co_await std::move(initiator); + auto sr_waiter = co_await std::move(waiter); + EXPECT_EQ(calls, 1); + EXPECT_EQ(sr_initiator.get(), sr_waiter.get()); +} + +TEST_CORO(FetchReadCoalescer, StaleResultReReads) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + // Read at hwm=100. + auto sr = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls, 7, model::offset{100})); + EXPECT_EQ(calls, 1); + + // Same bound is fresh -> shared. A grown bound is stale -> the reader must + // re-read. + auto fresh = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 1); + EXPECT_EQ(sr.get(), fresh.get()); + + auto grown = co_await c.get_or_insert( + key, model::offset{200}, counting_read(calls, 9, model::offset{200})); + EXPECT_EQ(calls, 2); + EXPECT_NE(sr.get(), grown.get()); +} + +TEST_CORO(FetchReadCoalescer, StaleEntryRegeneratesInPlace) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + auto first = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls, 7, model::offset{100})); + EXPECT_EQ(calls, 1); + + // Partition grew: the retained result is stale, so the reader re-reads and + // regenerates the entry in place with the grown result. + auto second = co_await c.get_or_insert( + key, model::offset{200}, counting_read(calls, 9, model::offset{200})); + EXPECT_EQ(calls, 2); + EXPECT_NE(first.get(), second.get()); + + // A reader at the new bound now ready-hits the regenerated result. + auto third = co_await c.get_or_insert( + key, model::offset{200}, counting_read(calls)); + EXPECT_EQ(calls, 2); + EXPECT_EQ(second.get(), third.get()); +} + +TEST_CORO(FetchReadCoalescer, ErrorResultNotRetained) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + auto sr = co_await c.get_or_insert( + key, model::offset{100}, counting_error(calls)); + // The error is handed to the initiator (and would be to waiters)... + EXPECT_EQ(sr->error, error_code::offset_out_of_range); + EXPECT_EQ(calls, 1); + + // ...but never retained: the next reader re-reads. + co_await c.get_or_insert(key, model::offset{100}, counting_error(calls)); + EXPECT_EQ(calls, 2); +} + +TEST_CORO(FetchReadCoalescer, EmptyResultNotRetained) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + auto sr = co_await c.get_or_insert( + key, model::offset{100}, counting_empty(calls)); + EXPECT_FALSE(sr->has_data()); + EXPECT_EQ(calls, 1); + + // An empty response drains instantly, so it isn't retained: re-read. + co_await c.get_or_insert(key, model::offset{100}, counting_empty(calls)); + EXPECT_EQ(calls, 2); +} + +TEST_CORO(FetchReadCoalescer, ReadThrowsPropagatesAndDoesNotRetain) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + bool threw = false; + try { + co_await c.get_or_insert(key, model::offset{100}, [&calls] { + ++calls; + return ss::make_exception_future( + std::runtime_error("boom")); + }); + } catch (const std::runtime_error&) { + threw = true; + } + EXPECT_TRUE(threw); + EXPECT_EQ(calls, 1); + + // A throw leaves nothing retained: the next reader re-reads. + co_await c.get_or_insert(key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 2); +} + +TEST_CORO(FetchReadCoalescer, DisableClearsCache) { + config::mock_property enabled{true}; + fetch_read_coalescer c(test_cache_config(), enabled.bind()); + auto key = make_key(); + + int calls = 0; + auto sr = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls)); + auto shared = co_await c.get_or_insert( + key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 1); + + enabled.update(false); + co_await ss::yield(); + EXPECT_FALSE(c.enabled()); + + enabled.update(true); + co_await ss::yield(); + EXPECT_TRUE(c.enabled()); + + // Re-enabling starts from an empty cache: even though sr/shared still hold + // the blob, the entry is gone, so the reader misses and re-reads. + co_await c.get_or_insert(key, model::offset{100}, counting_read(calls)); + EXPECT_EQ(calls, 2); +} + +} // namespace kafka From 8a888e9f03f073304a1c5ce3f07060b9c71bdcfc Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Thu, 2 Jul 2026 22:04:47 -0400 Subject: [PATCH 4/7] config: add kafka_fetch_read_coalescing_enabled Live (needs_restart::no) cluster flag, default off, gating the fetch read coalescer. Wired into the fetch path in a following commit. --- src/v/config/configuration.cc | 8 ++++++++ src/v/config/configuration.h | 1 + 2 files changed, 9 insertions(+) diff --git a/src/v/config/configuration.cc b/src/v/config/configuration.cc index 5dd2588b6467e..db637144f5fd1 100644 --- a/src/v/config/configuration.cc +++ b/src/v/config/configuration.cc @@ -899,6 +899,14 @@ configuration::configuration() "Use a separate scheduler group for fetch processing.", {.needs_restart = needs_restart::no, .visibility = visibility::tunable}, true) + , kafka_fetch_read_coalescing_enabled( + *this, + "kafka_fetch_read_coalescing_enabled", + "Coalesce concurrent fetches of the same partition offset into one read " + "and serialization shared across the requesting consumers, reducing " + "duplicate reads and fetch-response memory under high fanout.", + {.needs_restart = needs_restart::no, .visibility = visibility::tunable}, + false) , use_produce_scheduler_group( *this, "use_produce_scheduler_group", diff --git a/src/v/config/configuration.h b/src/v/config/configuration.h index 59b601a65ea45..3b9177a58427b 100644 --- a/src/v/config/configuration.h +++ b/src/v/config/configuration.h @@ -222,6 +222,7 @@ struct configuration final : public config_store { enum_property log_compression_type; property fetch_max_bytes; property use_fetch_scheduler_group; + property kafka_fetch_read_coalescing_enabled; property use_produce_scheduler_group; property use_kafka_handler_scheduler_group; property kafka_handler_latency_all; From 523b8e9107b4da93300ca610d65b36f4e12d7323 Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Thu, 2 Jul 2026 23:16:30 -0400 Subject: [PATCH 5/7] kafka: wire-in fetch_read_coalescer into fetch handler 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. --- src/v/kafka/server/handlers/fetch.cc | 130 +++++++++++++++++++------ src/v/kafka/server/handlers/fetch.h | 6 +- src/v/kafka/server/server.cc | 4 + src/v/kafka/server/server.h | 6 ++ src/v/kafka/server/tests/fetch_test.cc | 3 +- 5 files changed, 119 insertions(+), 30 deletions(-) diff --git a/src/v/kafka/server/handlers/fetch.cc b/src/v/kafka/server/handlers/fetch.cc index 5d0756d19bbe6..203760702c99e 100644 --- a/src/v/kafka/server/handlers/fetch.cc +++ b/src/v/kafka/server/handlers/fetch.cc @@ -21,6 +21,7 @@ #include "kafka/protocol/batch_consumer.h" #include "kafka/protocol/errors.h" #include "kafka/protocol/fetch.h" +#include "kafka/server/fetch_read_coalescer.h" #include "kafka/server/fetch_session.h" #include "kafka/server/fwd.h" #include "kafka/server/handlers/details/leader_epoch.h" @@ -205,6 +206,61 @@ static ss::future read_from_partition( std::move(aborted_transactions)); } +// Clone a shared read into a fresh read_result for one response. Share the +// iobuf and copy the metadata. memory_units is left unset; the caller installs +// an aliasing pointer that co-owns the shared result. +static read_result clone_read_result(const read_result& src) { + if (src.error != error_code::none) { + return read_result( + src.error, + src.start_offset, + src.high_watermark, + src.last_stable_offset); + } + return read_result( + src.has_data() ? std::make_unique(src.data->share()) : nullptr, + src.start_offset, + src.data_base_offset, + src.data_last_offset, + src.batch_count, + src.high_watermark, + src.last_stable_offset, + src.delta_from_tip_ms, + src.aborted_transactions); +} + +// Allocate memory units, read, and adjust the reservation to the result size. +static ss::future read_with_units( + fetch_memory_units_manager& units_mgr, + kafka::partition_proxy part, + ntp_fetch_config& ntp_config, + model::offset lso, + std::optional deadline, + bool obligatory_batch_read) { + auto memory_units = units_mgr.zero_units(); + if (!ntp_config.cfg.skip_read) { + memory_units = units_mgr.allocate_memory_units( + ntp_config.ktp(), + ntp_config.cfg.max_bytes, + ntp_config.cfg.max_batch_size, + ntp_config.cfg.avg_batch_size, + obligatory_batch_read); + if (!memory_units.has_units()) { + ntp_config.cfg.skip_read = true; + } else if (ntp_config.cfg.max_bytes > memory_units.num_units()) { + ntp_config.cfg.max_bytes = memory_units.num_units(); + } + } + auto result = co_await read_from_partition( + std::move(part), lso, ntp_config.cfg, deadline); + // Units can be increased here too: an obligatory read has no strict limit + // and may return a batch larger than the reserved size. + memory_units.adjust_units(result.data_size_bytes()); + result.memory_units = std::make_shared( + std::move(memory_units)); + co_return result; +} + /** * Entry point for reading from an ntp. This is executed on NTP home core and * build error responses if anything goes wrong. @@ -216,7 +272,8 @@ static ss::future do_read_from_ntp( ntp_fetch_config ntp_config, std::optional deadline, const bool obligatory_batch_read, - fetch_memory_units_manager& units_mgr) { + fetch_memory_units_manager& units_mgr, + fetch_read_coalescer& coalescer) { /* * lookup the ntp's partition */ @@ -298,32 +355,44 @@ static ss::future do_read_from_ntp( // is larger. This is needed to conform with KIP-74. ntp_config.cfg.strict_max_bytes = !obligatory_batch_read; - // control available memory - auto memory_units = units_mgr.zero_units(); - if (!ntp_config.cfg.skip_read) { - memory_units = units_mgr.allocate_memory_units( - ntp_config.ktp(), - ntp_config.cfg.max_bytes, - ntp_config.cfg.max_batch_size, - ntp_config.cfg.avg_batch_size, + auto do_read = [&] { + return read_with_units( + units_mgr, + std::move(*kafka_partition), + ntp_config, + maybe_lso.value(), + deadline, obligatory_batch_read); - if (!memory_units.has_units()) { - ntp_config.cfg.skip_read = true; - } else if (ntp_config.cfg.max_bytes > memory_units.num_units()) { - ntp_config.cfg.max_bytes = memory_units.num_units(); - } + }; + + if (!coalescer.enabled()) { + co_return co_await do_read(); } - auto result = co_await read_from_partition( - std::move(*kafka_partition), maybe_lso.value(), ntp_config.cfg, deadline); + auto get_result = [&](const shared_read& sr) { + read_result r = clone_read_result(*sr); + // The units are held until the response is sent, so co-owning the + // cached read through them (aliasing ctor) keeps it alive until then. + if (sr->memory_units) { + r.memory_units = std::shared_ptr( + sr, sr->memory_units.get()); + } + return r; + }; - // Note that units can be both increased and decreassed here. Increases - // happen because there is no strict limit on read size when reading the - // obligatory batch. - memory_units.adjust_units(result.data_size_bytes()); - result.memory_units = std::make_shared( - std::move(memory_units)); - co_return result; + auto current_bound = ntp_config.cfg.isolation_level + == model::isolation_level::read_committed + ? maybe_lso.value() + : kafka_partition->high_watermark(); + coalesce_key key{ + .ktp = ntp_config.ktp_with_hash(), + .fetch_offset = ntp_config.cfg.start_offset, + .level = ntp_config.cfg.isolation_level, + .max_bytes = ntp_config.cfg.max_bytes, + .obligatory = obligatory_batch_read}; + + co_return get_result( + co_await coalescer.get_or_insert(key, current_bound, std::move(do_read))); } namespace testing { @@ -336,7 +405,8 @@ ss::future read_from_ntp( fetch_config config, std::optional deadline, const bool obligatory_batch_read, - fetch_memory_units_manager& units_mgr) { + fetch_memory_units_manager& units_mgr, + fetch_read_coalescer& coalescer) { return do_read_from_ntp( cluster_pm, md_cache, @@ -344,7 +414,8 @@ ss::future read_from_ntp( {{ktp.get_topic(), ktp.get_partition()}, std::move(config)}, deadline, obligatory_batch_read, - units_mgr); + units_mgr, + coalescer); } } // namespace testing @@ -476,7 +547,8 @@ static ss::future> fetch_ntps( std::optional deadline, model::timeout_clock::time_point fetch_deadline, const size_t bytes_left, - fetch_memory_units_manager& units_mgr) { + fetch_memory_units_manager& units_mgr, + fetch_read_coalescer& coalescer) { size_t total_read_size = 0; // bytes_left comes from the fetch plan and also accounts for the max_bytes @@ -529,7 +601,8 @@ static ss::future> fetch_ntps( ntp_cfg, fetch_deadline, obligatory_batch_read, - units_mgr) + units_mgr, + coalescer) .then([&, cfg_idx](read_result&& res) { res.partition = ntp_cfg.ktp().get_partition(); @@ -715,7 +788,8 @@ class fetch_worker { _ctx.deadline, _ctx.fetch_deadline, _ctx.bytes_left, - _ctx.srv.fetch_units_manager()); + _ctx.srv.fetch_units_manager(), + _ctx.srv.read_coalescer()); // If we weren't able to read the last_visible_index for a partition // before calling `fetch_ntps_in_parallel` then we need to diff --git a/src/v/kafka/server/handlers/fetch.h b/src/v/kafka/server/handlers/fetch.h index a5f9862b81dde..12a550cea7f29 100644 --- a/src/v/kafka/server/handlers/fetch.h +++ b/src/v/kafka/server/handlers/fetch.h @@ -28,6 +28,8 @@ namespace kafka { +class fetch_read_coalescer; + std::optional fetch_scheduling_group_provider(const connection_context&); @@ -270,6 +272,7 @@ struct ntp_fetch_config { fetch_config cfg; const model::ktp& ktp() const { return _ktp; } + const model::ktp_with_hash& ktp_with_hash() const { return _ktp; } fmt::iterator format_to(fmt::iterator it) const { return fmt::format_to(it, R"({{"{}": {}}})", ktp(), cfg); @@ -453,7 +456,8 @@ ss::future read_from_ntp( fetch_config, std::optional, bool obligatory_batch_read, - fetch_memory_units_manager& units_mgr); + fetch_memory_units_manager& units_mgr, + fetch_read_coalescer& coalescer); /** * Create a fetch plan with the simple fetch planner. diff --git a/src/v/kafka/server/server.cc b/src/v/kafka/server/server.cc index fa4072957de7a..70a3a6fc16786 100644 --- a/src/v/kafka/server/server.cc +++ b/src/v/kafka/server/server.cc @@ -211,6 +211,10 @@ server::server( return container().local().fetch_units_manager(); }, config::shard_local_cfg().kafka_max_message_size_upper_limit_bytes.bind()) + , _fetch_read_coalescer( + // ~0.5 MB/shard when coalescing is enabled. + {.cache_size = 1024, .small_size = 128}, + config::shard_local_cfg().kafka_fetch_read_coalescing_enabled.bind()) , _probe(std::make_unique()) , _sasl_probe(std::make_unique()) , _read_dist_probe(std::make_unique()) diff --git a/src/v/kafka/server/server.h b/src/v/kafka/server/server.h index a125de1e54683..f68a6351ea317 100644 --- a/src/v/kafka/server/server.h +++ b/src/v/kafka/server/server.h @@ -21,6 +21,7 @@ #include "kafka/server/fetch_memory_units.h" #include "kafka/server/fetch_metadata_cache.h" #include "kafka/server/fetch_pid_controller.h" +#include "kafka/server/fetch_read_coalescer.h" #include "kafka/server/fetch_session_cache.h" #include "kafka/server/fwd.h" #include "kafka/server/handlers/fetch/replica_selector.h" @@ -239,6 +240,10 @@ class server final return _fetch_units_manager; } + fetch_read_coalescer& read_coalescer() noexcept { + return _fetch_read_coalescer; + } + ss::future<> revoke_credentials(std::string_view name); // Returns a default scheduling group that is intended to be used for @@ -311,6 +316,7 @@ class server final security::krb5::configurator _krb_configurator; ssx::semaphore _memory_fetch_sem; fetch_memory_units_manager _fetch_units_manager; + fetch_read_coalescer _fetch_read_coalescer; handler_probe_manager _handler_probes; metrics::internal_metric_groups _metrics; diff --git a/src/v/kafka/server/tests/fetch_test.cc b/src/v/kafka/server/tests/fetch_test.cc index 08026580587ef..bd48b422fc5da 100644 --- a/src/v/kafka/server/tests/fetch_test.cc +++ b/src/v/kafka/server/tests/fetch_test.cc @@ -186,7 +186,8 @@ FIXTURE_TEST(read_from_ntp_max_bytes, redpanda_thread_fixture) { config, model::no_timeout, false, - octx.rctx.server().local().fetch_units_manager()); + octx.rctx.server().local().fetch_units_manager(), + octx.rctx.server().local().read_coalescer()); }) .get(); BOOST_TEST_REQUIRE(res.has_data()); From 7319befa041417d89dd39b8d5424811faeb66219 Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Fri, 3 Jul 2026 00:30:17 -0400 Subject: [PATCH 6/7] rptest: add fetch read coalescing fanout test 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. --- .../tests/fetch_read_coalescing_test.py | 113 ++++++++++++++++++ .../type-checking/type-check-strictness.json | 1 + 2 files changed, 114 insertions(+) create mode 100644 tests/rptest/tests/fetch_read_coalescing_test.py diff --git a/tests/rptest/tests/fetch_read_coalescing_test.py b/tests/rptest/tests/fetch_read_coalescing_test.py new file mode 100644 index 0000000000000..590fe027f936e --- /dev/null +++ b/tests/rptest/tests/fetch_read_coalescing_test.py @@ -0,0 +1,113 @@ +# Copyright 2026 Redpanda Data, Inc. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.md +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0 + +import time + +from ducktape.utils.util import wait_until + +from rptest.clients.rpk import RpkTool +from rptest.clients.types import TopicSpec +from rptest.services.cluster import cluster +from rptest.services.redpanda import MetricsEndpoint +from rptest.services.rpk_consumer import RpkConsumer +from rptest.tests.redpanda_test import RedpandaTest + +RECORD_COUNT = 50 +# One node per consumer; the fanout is the thundering herd the coalescer exists +# for. +FANOUT = 6 + + +class FetchReadCoalescingTest(RedpandaTest): + """End-to-end check for INC-2843 P2a fetch read coalescing: a fanout of + consumers reading the same partition offset shares one serialized read + instead of reading (and serializing) the partition once per consumer, while + every consumer still receives the correct data.""" + + topics = (TopicSpec(name="coalesce", partition_count=1, replication_factor=1),) + + def __init__(self, test_ctx, *args, **kwargs): + super().__init__(test_ctx, num_brokers=1, *args, **kwargs) + + def _coalescer_metric(self, name: str) -> int: + return int( + self.redpanda.metric_sum( + f"vectorized_kafka_fetch_read_coalescer_{name}_total", + metrics_endpoint=MetricsEndpoint.METRICS, + expect_metric=True, + ) + ) + + @cluster(num_nodes=1 + FANOUT) + def test_fanout_shares_one_read(self): + rpk = RpkTool(self.redpanda) + + # needs_restart::no, so coalescing turns on without a bounce. + self.redpanda.set_cluster_config({"kafka_fetch_read_coalescing_enabled": True}) + + # Independent consumers (no shared group), all reading partition 0 from + # the start. Started against the still-empty topic so they park in a + # long-poll on the same partition; the produce below then wakes the + # whole fleet on each record, so every offset is fetched near + # simultaneously by all of them and coalesces. + consumers = [ + RpkConsumer( + self.test_context, + self.redpanda, + self.topic, + partitions=[0], + offset="oldest", + num_msgs=RECORD_COUNT, + ) + for _ in range(FANOUT) + ] + for c in consumers: + c.start() + # Let the fleet connect and park before the first record arrives. + time.sleep(5) + + expected = [(f"key-{i:04d}", f"val-{i:04d}") for i in range(RECORD_COUNT)] + for key, value in expected: + rpk.produce(self.topic, key, value, partition=0) + + for i, c in enumerate(consumers): + wait_until( + lambda c=c: c.message_count >= RECORD_COUNT, + timeout_sec=60, + backoff_sec=0.5, + err_msg=f"consumer {i} got {c.message_count}/{RECORD_COUNT}", + ) + for c in consumers: + c.stop() + + # Correctness: serving a shared read must deliver each consumer the same + # bytes a solo read would have. + for i, c in enumerate(consumers): + got = [ + (m["key"], m["value"]) + for m in sorted(c.messages, key=lambda m: int(m["offset"])) + ][:RECORD_COUNT] + assert got == expected, ( + f"consumer {i} data mismatch (first records: {got[:2]})" + ) + + insertions = self._coalescer_metric("insertions") + reinsertions = self._coalescer_metric("reinsertions") + ready_hits = self._coalescer_metric("ready_hits") + inflight_hits = self._coalescer_metric("inflight_hits") + self.logger.info( + f"coalescer: insertions={insertions} reinsertions={reinsertions} " + f"ready_hits={ready_hits} inflight_hits={inflight_hits}" + ) + + # Coalescing engaged: some of the fanout's reads were served from a + # shared result rather than each reading the partition afresh. + assert ready_hits + inflight_hits > 0, ( + "expected the fanout to coalesce reads, but saw no hits" + ) diff --git a/tools/type-checking/type-check-strictness.json b/tools/type-checking/type-check-strictness.json index 67fca00acd5eb..218b8765db384 100644 --- a/tools/type-checking/type-check-strictness.json +++ b/tools/type-checking/type-check-strictness.json @@ -142,6 +142,7 @@ "rptest/tests/fetch_after_deletion_test.py", "rptest/tests/fetch_fairness_test.py", "rptest/tests/fetch_long_poll_test.py", + "rptest/tests/fetch_read_coalescing_test.py", "rptest/tests/follower_fetching_test.py", "rptest/tests/host_metrics_test.py", "rptest/tests/idempotency_test.py", From 5f1e1ab969f98c9f6c80d006437cb9934963e4d0 Mon Sep 17 00:00:00 2001 From: Brandon Allard Date: Fri, 3 Jul 2026 01:24:45 -0400 Subject: [PATCH 7/7] tmp: change kafka_fetch_read_coalescing_enabled default to true This commit is just to run the full CI against this config enabled. Will be dropped before merge. --- src/v/config/configuration.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/v/config/configuration.cc b/src/v/config/configuration.cc index db637144f5fd1..6971fa53843d6 100644 --- a/src/v/config/configuration.cc +++ b/src/v/config/configuration.cc @@ -906,7 +906,7 @@ configuration::configuration() "and serialization shared across the requesting consumers, reducing " "duplicate reads and fetch-response memory under high fanout.", {.needs_restart = needs_restart::no, .visibility = visibility::tunable}, - false) + true) , use_produce_scheduler_group( *this, "use_produce_scheduler_group",