Add serialized response coalescing for high fanout fetch workloads

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},
+      true)
   , use_produce_scheduler_group(
       *this,
       "use_produce_scheduler_group",

src/v/config/configuration.h

@@ -222,6 +222,7 @@ struct configuration final : public config_store {
     enum_property<model::compression> log_compression_type;
     property<size_t> fetch_max_bytes;
     property<bool> use_fetch_scheduler_group;
+    property<bool> kafka_fetch_read_coalescing_enabled;
     property<bool> use_produce_scheduler_group;
     property<bool> use_kafka_handler_scheduler_group;
     property<bool> kafka_handler_latency_all;

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",

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<model::ktp_with_hash>{}(k.ktp);
+    hash::combine(
+      h,
+      k.fetch_offset(),
+      static_cast<int8_t>(k.level),
+      k.max_bytes,
+      k.obligatory);
+    return h;
+}
+
+fetch_read_coalescer::fetch_read_coalescer(
+  cache_config cfg, config::binding<bool> 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<shared_read> 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<coalesce_entry> 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<shared_read>(std::move(r));
+        }
+        ++_reinsertions;
+    } else {
+        entry = ss::make_shared<coalesce_entry>();
+        _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<shared_read> fetch_read_coalescer::read_and_publish(
+  ss::shared_ptr<coalesce_entry> entry, read_fn fn) {
+    entry->begin_read();
+    try {
+        auto shared = std::make_shared<const read_result>(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

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 <seastar/core/future.hh>
+#include <seastar/core/shared_future.hh>
+#include <seastar/util/noncopyable_function.hh>
+
+#include <memory>
+#include <optional>
+
+namespace kafka {
+
+// Defined in kafka/server/handlers/fetch.h.
+struct read_result;
+
+using shared_read = std::shared_ptr<const read_result>;
+
+/// 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<ss::shared_promise<shared_read>> inflight;
+        // The last completed read, held weakly so the coalescer never pins it.
+        std::weak_ptr<const read_result> 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<coalesce_key, coalesce_entry, coalesce_key_hash>;
+    using cache_config = cache_t::config;
+
+    /// Always constructed; disabling clears the cache via the binding watch.
+    fetch_read_coalescer(cache_config, config::binding<bool> 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<ss::future<read_result>()>;
+
+    /// 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<shared_read>
+    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<shared_read>
+      read_and_publish(ss::shared_ptr<coalesce_entry>, read_fn);
+
+    void on_enabled_change();
+    void setup_metrics();
+
+    cache_config _cache_config;
+    config::binding<bool> _enabled;
+    // Held via optional because chunked_kv_cache has no clear() and deletes its
+    // move; disable reset()s it.
+    std::optional<cache_t> _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