diff --git a/src/v/config/configuration.cc b/src/v/config/configuration.cc
index 5dd2588b6467e..6971fa53843d6 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},
+      true)
   , 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<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;
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_memory_units.h b/src/v/kafka/server/fetch_memory_units.h
index 004688d693c4b..968e96b6ac275 100644
--- a/src/v/kafka/server/fetch_memory_units.h
+++ b/src/v/kafka/server/fetch_memory_units.h
@@ -19,6 +19,9 @@
 #include <seastar/core/gate.hh>
 #include <seastar/core/sharded.hh>
 
+#include <memory>
+#include <variant>
+
 namespace kafka {
 
 using namespace std::chrono_literals;
@@ -168,4 +171,51 @@ class [[nodiscard]] fetch_memory_units {
     fetch_memory_units_manager::local_instance_fn _local_instance_fn;
 };
 
+/// Holds the memory units reserved for a read's data until its response is
+/// sent. A non-coalesced read owns its units inline, with no allocation. A
+/// coalesced read is shared by several responses, so each response instead
+/// holds a shared handle that co-owns the shared read result (and thus its
+/// inline units) until that response is sent.
+class fetch_units_holder {
+public:
+    using shared_units = std::shared_ptr<const fetch_memory_units>;
+
+    fetch_units_holder() noexcept = default;
+    explicit fetch_units_holder(fetch_memory_units units) noexcept
+      : _units(std::move(units)) {}
+    explicit fetch_units_holder(shared_units units) noexcept
+      : _units(std::move(units)) {}
+
+    fetch_units_holder(fetch_units_holder&&) noexcept = default;
+    fetch_units_holder& operator=(fetch_units_holder&&) noexcept = default;
+    fetch_units_holder(const fetch_units_holder&) = delete;
+    fetch_units_holder& operator=(const fetch_units_holder&) = delete;
+    ~fetch_units_holder() noexcept = default;
+
+    size_t num_units() const {
+        if (const auto* u = std::get_if<fetch_memory_units>(&_units)) {
+            return u->num_units();
+        }
+        if (const auto* u = std::get_if<shared_units>(&_units)) {
+            return *u ? (*u)->num_units() : 0;
+        }
+        return 0;
+    }
+
+    /// Pointer to the held units, or nullptr if none. A coalesced clone aliases
+    /// this to co-own the shared read result it was cloned from.
+    const fetch_memory_units* get() const {
+        if (const auto* u = std::get_if<fetch_memory_units>(&_units)) {
+            return u;
+        }
+        if (const auto* u = std::get_if<shared_units>(&_units)) {
+            return u->get();
+        }
+        return nullptr;
+    }
+
+private:
+    std::variant<std::monostate, fetch_memory_units, shared_units> _units;
+};
+
 } // namespace kafka
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..aa47104c6b950
--- /dev/null
+++ b/src/v/kafka/server/fetch_read_coalescer.cc
@@ -0,0 +1,189 @@
+/*
+ * 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"
+#include "ssx/future-util.h"
+
+#include <memory>
+
+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_factory make_read) {
+    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;
+    }
+    // When the key doesn't exist or has expired drive the read in the
+    // background so it outlives any single waiter. Every caller waits the
+    // returned shared future (bounded by its own deadline at the call site).
+    return start_read(std::move(entry), std::move(make_read));
+}
+
+ss::future<shared_read> fetch_read_coalescer::start_read(
+  ss::shared_ptr<coalesce_entry> entry, read_factory make_read) {
+    if (_read_gate.is_closed()) {
+        return ss::make_exception_future<shared_read>(
+          ss::gate_closed_exception());
+    }
+
+    entry->begin_read();
+    // A synchronously-completing read resets the in-flight promise before
+    // spawn_with_gate returns. So save a reference to it to return.
+    auto fut = entry->inflight->get_shared_future();
+    read_fn fn = make_read(_read_abort, read_deadline());
+    ssx::spawn_with_gate(_read_gate, [entry, fn = std::move(fn)]() mutable {
+        return read_and_publish(std::move(entry), std::move(fn));
+    });
+    return fut;
+}
+
+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<coalesce_entry> entry, read_fn fn) {
+    try {
+        auto shared = std::make_shared<const read_result>(co_await fn());
+        entry->finish_read(std::move(shared));
+    } catch (...) {
+        entry->fail_read(std::current_exception());
+    }
+}
+
+ss::future<> fetch_read_coalescer::stop() {
+    _read_abort.request_abort();
+    return _read_gate.close();
+}
+
+model::timeout_clock::time_point fetch_read_coalescer::read_deadline() const {
+    return model::timeout_clock::now()
+           + config::shard_local_cfg().kafka_fetch_request_timeout_ms();
+}
+
+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..c25dc03776f6e
--- /dev/null
+++ b/src/v/kafka/server/fetch_read_coalescer.h
@@ -0,0 +1,173 @@
+/*
+ * 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 "model/timeout_clock.h"
+#include "utils/chunked_kv_cache.h"
+
+#include <seastar/core/abort_source.hh>
+#include <seastar/core/future.hh>
+#include <seastar/core/gate.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>()>;
+    /// Produces the read for a miss. Invoked at most once per get_or_insert,
+    /// only on the miss path, so a hit never pays to build the read. The
+    /// coalescer passes it the neutral abort source and deadline the read must
+    /// run under, so the resulting read_fn does not depend on any single caller
+    /// and can safely outlive its creator.
+    using read_factory = ss::noncopyable_function<read_fn(
+      ss::abort_source&, model::timeout_clock::time_point)>;
+
+    /// 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 the factory produces a read that is started in the
+    ///   background and its result is shared and retained.
+    /// The read runs at most once, only on the miss path, driven by the
+    /// coalescer's gate so it outlives any single waiter. 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). Callers should
+    /// bound their wait on the returned future by their own deadline; giving up
+    /// early does not disturb the in-flight read serving other waiters.
+    ss::future<shared_read> get_or_insert(
+      const coalesce_key&, model::offset current_bound, read_factory);
+
+    /// Drains in-flight background reads. Must be awaited before the
+    /// shard-local services the reads depend on (e.g. the memory units manager)
+    /// are stopped.
+    ss::future<> stop();
+
+private:
+    // Materialize the read from the factory (bound to the neutral abort and
+    // deadline), install the in-flight promise, and spawn read_and_publish on
+    // the gate, returning the shared future callers wait on.
+    ss::future<shared_read>
+      start_read(ss::shared_ptr<coalesce_entry>, read_factory);
+    // Run read_fn on `entry` and share its result with the entry's waiters,
+    // retaining it iff it is data-bearing and error-free. Runs as a background
+    // gate task, so it never throws. Read errors reach waiters via the
+    // in-flight promise.
+    static ss::future<>
+      read_and_publish(ss::shared_ptr<coalesce_entry>, read_fn);
+    // The neutral time bound a coalesced read runs under.
+    model::timeout_clock::time_point read_deadline() const;
+
+    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;
+    // Fired on stop() so a slow background read (e.g. cloud storage) is cut
+    // short rather than waited out; the gate then drains the aborted reads.
+    ss::abort_source _read_abort;
+    ss::gate _read_gate;
+
+    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/handlers/fetch.cc b/src/v/kafka/server/handlers/fetch.cc
index 7a5a7827ab21f..f06c1904277f5 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"
@@ -46,6 +47,7 @@
 #include <seastar/core/scheduling.hh>
 #include <seastar/core/sleep.hh>
 #include <seastar/core/timer.hh>
+#include <seastar/core/with_timeout.hh>
 #include <seastar/util/log.hh>
 
 #include <boost/range/irange.hpp>
@@ -106,7 +108,8 @@ static ss::future<read_result> read_from_partition(
   kafka::partition_proxy part,
   model::offset lso,
   fetch_config config,
-  std::optional<model::timeout_clock::time_point> deadline) {
+  std::optional<model::timeout_clock::time_point> deadline,
+  model::opt_abort_source_t abort) {
     auto hw = part.high_watermark();
     auto start_o = part.start_offset();
     // if we have no data read, return fast
@@ -122,9 +125,7 @@ static ss::future<read_result> read_from_partition(
       0,
       config.max_bytes,
       std::nullopt,
-      config.abort_source.has_value()
-        ? config.abort_source.value().get().local()
-        : model::opt_abort_source_t{},
+      abort,
       config.client_address,
       config.strict_max_bytes);
 
@@ -205,24 +206,38 @@ static ss::future<read_result> read_from_partition(
       std::move(aborted_transactions));
 }
 
-/**
- * Entry point for reading from an ntp. This is executed on NTP home core and
- * build error responses if anything goes wrong.
- */
-static ss::future<read_result> do_read_from_ntp(
-  cluster::partition_manager& cluster_pm,
-  const cluster::metadata_cache& md_cache,
-  const replica_selector& replica_selector,
-  ntp_fetch_config ntp_config,
-  std::optional<model::timeout_clock::time_point> 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;
+// 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<iobuf>(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);
+}
 
-    // control available memory
+// Allocate memory units, read, and adjust the reservation to the result size.
+static ss::future<read_result> read_with_units(
+  fetch_memory_units_manager& units_mgr,
+  kafka::partition_proxy part,
+  ntp_fetch_config& ntp_config,
+  model::offset lso,
+  std::optional<model::timeout_clock::time_point> deadline,
+  bool obligatory_batch_read,
+  model::opt_abort_source_t abort) {
     auto memory_units = units_mgr.zero_units();
     if (!ntp_config.cfg.skip_read) {
         memory_units = units_mgr.allocate_memory_units(
@@ -237,7 +252,28 @@ static ss::future<read_result> do_read_from_ntp(
             ntp_config.cfg.max_bytes = memory_units.num_units();
         }
     }
+    auto result = co_await read_from_partition(
+      std::move(part), lso, ntp_config.cfg, deadline, abort);
+    // 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 = fetch_units_holder{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.
+ */
+static ss::future<read_result> do_read_from_ntp(
+  cluster::partition_manager& cluster_pm,
+  const cluster::metadata_cache& md_cache,
+  const replica_selector& replica_selector,
+  ntp_fetch_config ntp_config,
+  std::optional<model::timeout_clock::time_point> deadline,
+  const bool obligatory_batch_read,
+  fetch_memory_units_manager& units_mgr,
+  fetch_read_coalescer& coalescer) {
     /*
      * lookup the ntp's partition
      */
@@ -314,15 +350,114 @@ static ss::future<read_result> do_read_from_ntp(
               preferred_replica);
         }
     }
-    auto result = co_await read_from_partition(
-      std::move(*kafka_partition), maybe_lso.value(), ntp_config.cfg, deadline);
+    // 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;
 
-    // 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::move(memory_units);
-    co_return result;
+    // The caller's own abort source, if any. Bounds its inline read (disabled)
+    // or its wait on the shared read (coalesced); it is not passed to the
+    // coalesced read itself, which uses the coalescer's neutral abort.
+    const auto caller_abort
+      = ntp_config.cfg.abort_source.has_value()
+          ? model::opt_abort_source_t{ntp_config.cfg.abort_source.value()
+                                        .get()
+                                        .local()}
+          : model::opt_abort_source_t{};
+
+    if (!coalescer.enabled()) {
+        co_return co_await read_with_units(
+          units_mgr,
+          std::move(*kafka_partition),
+          ntp_config,
+          maybe_lso.value(),
+          deadline,
+          obligatory_batch_read,
+          caller_abort);
+    }
+
+    // Snapshot offsets before the proxy is moved into the read, so a caller
+    // that times out early can still build an empty result.
+    const auto start_o = kafka_partition->start_offset();
+    const auto hwm = kafka_partition->high_watermark();
+    const auto lso = maybe_lso.value();
+    const auto current_bound = ntp_config.cfg.isolation_level
+                                   == model::isolation_level::read_committed
+                                 ? lso
+                                 : hwm;
+    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};
+
+    // The coalescer drives this read in the background and shares it with every
+    // waiter, so it must not depend on any single caller. The factory runs only
+    // on a miss (a hit never pays for it): it copies the config and owns the
+    // proxy, and the coalescer hands it the neutral abort and deadline the read
+    // runs under. Dropping the per-request abort keeps a caller's disconnect
+    // from cutting the shared read short.
+    auto make_read = [&units_mgr,
+                      &ntp_config,
+                      part = std::move(*kafka_partition),
+                      lso,
+                      obligatory_batch_read](
+                       ss::abort_source& read_abort,
+                       model::timeout_clock::time_point read_deadline) mutable
+      -> fetch_read_coalescer::read_fn {
+        return [&units_mgr,
+                part = std::move(part),
+                cfg = ntp_config,
+                lso,
+                read_deadline,
+                obligatory_batch_read,
+                abort = model::opt_abort_source_t{read_abort}]() mutable {
+            cfg.cfg.abort_source = std::nullopt;
+            return read_with_units(
+              units_mgr,
+              std::move(part),
+              cfg,
+              lso,
+              read_deadline,
+              obligatory_batch_read,
+              abort);
+        };
+    };
+
+    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 (const auto* units = sr->memory_units.get()) {
+            r.memory_units = fetch_units_holder{
+              std::shared_ptr<const fetch_memory_units>(sr, units)};
+        }
+        return r;
+    };
+
+    // Wait on the shared read only until this caller's own deadline or its own
+    // request abort. Either means this caller gave up waiting, not that the
+    // read failed, so we return empty for this partition (its next fetch picks
+    // up the retained result) while the read keeps running for other waiters.
+    auto shared = coalescer.get_or_insert(
+      key, current_bound, std::move(make_read));
+    const auto wait_deadline = deadline.value_or(
+      model::timeout_clock::time_point::max());
+    try {
+        shared_read sr;
+        if (caller_abort.has_value()) {
+            sr = co_await ssx::with_timeout_abortable(
+              std::move(shared), wait_deadline, caller_abort.value().get());
+        } else {
+            sr = co_await ss::with_timeout(wait_deadline, std::move(shared));
+        }
+        co_return get_result(sr);
+    } catch (const ss::timed_out_error&) {
+        co_return read_result(start_o, hwm, lso);
+    } catch (const ss::abort_requested_exception&) {
+        co_return read_result(start_o, hwm, lso);
+    }
 }
 
 namespace testing {
@@ -335,7 +470,8 @@ ss::future<read_result> read_from_ntp(
   fetch_config config,
   std::optional<model::timeout_clock::time_point> 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,
@@ -343,7 +479,8 @@ ss::future<read_result> read_from_ntp(
       {{ktp.get_topic(), ktp.get_partition()}, std::move(config)},
       deadline,
       obligatory_batch_read,
-      units_mgr);
+      units_mgr,
+      coalescer);
 }
 
 } // namespace testing
@@ -414,7 +551,7 @@ static void fill_fetch_responses(
             resp.preferred_read_replica = *res.preferred_replica;
         }
 
-        std::optional<fetch_memory_units> resp_units{};
+        fetch_units_holder resp_units;
         auto current_response_size = resp_it->response_size();
         auto bytes_left = octx.bytes_left - current_response_size;
 
@@ -475,7 +612,8 @@ static ss::future<chunked_vector<read_result>> fetch_ntps(
   std::optional<model::timeout_clock::time_point> 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
@@ -528,7 +666,8 @@ static ss::future<chunked_vector<read_result>> 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();
 
@@ -714,7 +853,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
@@ -1750,10 +1890,10 @@ ss::future<response_ptr> op_context::send_error_response(error_code ec) && {
 }
 
 ss::deleter op_context::response_memory_units_deleter() {
-    chunked_vector<fetch_memory_units> mu;
+    chunked_vector<fetch_units_holder> 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 +1915,7 @@ op_context::response_placeholder::response_placeholder(
 
 void op_context::response_placeholder::set(
   fetch_response::partition_response&& response,
-  std::optional<fetch_memory_units>&& response_memory_units) {
+  fetch_units_holder&& response_memory_units) {
     vassert(
       response.partition_index == _it->partition_response->partition_index,
       "Response and current partition ids have to be the same. Current "
@@ -1784,7 +1924,7 @@ void op_context::response_placeholder::set(
       response.partition_index);
     dassert(
       (response.records ? response.records->size_bytes() : 0)
-        == (response_memory_units ? response_memory_units->num_units() : 0),
+        == response_memory_units.num_units(),
       "Response units should equal the number of bytes in the response its "
       "self.");
 
diff --git a/src/v/kafka/server/handlers/fetch.h b/src/v/kafka/server/handlers/fetch.h
index 2a92dd1f64f37..022db50f46d7d 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<ss::scheduling_group>
 fetch_scheduling_group_provider(const connection_context&);
 
@@ -50,9 +52,7 @@ struct op_context {
     public:
         response_placeholder(fetch_response::iterator, op_context* ctx);
 
-        void set(
-          fetch_response::partition_response&&,
-          std::optional<fetch_memory_units>&&);
+        void set(fetch_response::partition_response&&, fetch_units_holder&&);
 
         const model::topic& topic() { return _it->partition->topic; }
         model::partition_id partition_id() {
@@ -85,18 +85,16 @@ struct op_context {
 
         // Returns the number of memory units held for this ntp.
         size_t num_memory_units() const {
-            return _response_memory_units ? _response_memory_units->num_units()
-                                          : 0;
+            return _response_memory_units.num_units();
         }
 
         // Adds/replaces the memory units that are held for this ntp.
-        void
-        replace_or_add_memory_units(std::optional<fetch_memory_units>&& units) {
+        void replace_or_add_memory_units(fetch_units_holder&& units) {
             _response_memory_units = std::move(units);
         }
 
         // Releases/returns the memory units that are held for this ntp.
-        std::optional<fetch_memory_units> release_memory_units() {
+        fetch_units_holder release_memory_units() {
             return std::move(_response_memory_units);
         }
 
@@ -104,8 +102,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<fetch_memory_units> _response_memory_units;
+        // Tracks memory used by response data in `_it`. A coalesced read result
+        // hands each of its responses a shared handle to the same units.
+        fetch_units_holder _response_memory_units;
     };
 
     using iteration_order_t
@@ -269,6 +268,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);
@@ -367,7 +367,7 @@ struct read_result {
     error_code error;
     model::partition_id partition;
     std::vector<cluster::tx::tx_range> aborted_transactions;
-    std::optional<fetch_memory_units> memory_units;
+    fetch_units_holder memory_units;
 };
 // struct aggregating fetch requests and corresponding response iterators for
 // the same shard
@@ -452,7 +452,8 @@ ss::future<read_result> read_from_ntp(
   fetch_config,
   std::optional<model::timeout_clock::time_point>,
   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..9ca8c107c588e 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<class kafka_probe>())
   , _sasl_probe(std::make_unique<class sasl_probe>())
   , _read_dist_probe(std::make_unique<read_distribution_probe>())
@@ -235,6 +239,8 @@ server::server(
 }
 ss::future<> server::stop() {
     co_await net::server::stop();
+    // Drain coalesced background reads before the units manager they depend on.
+    co_await _fetch_read_coalescer.stop();
     co_await _fetch_units_manager.stop();
 }
 
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/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..679fa770260bf
--- /dev/null
+++ b/src/v/kafka/server/tests/fetch_read_coalescer_test.cc
@@ -0,0 +1,334 @@
+/*
+ * 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 <seastar/core/coroutine.hh>
+#include <seastar/core/future.hh>
+
+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<char>(bytes));
+    return read_result(
+      std::make_unique<iobuf>(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. Tests pass read_factories; the
+// factory (and thus the read_fn it makes) runs only on a miss.
+
+// Wrap a single-use read_fn as a read_factory. Tests don't exercise the
+// coalescer's neutral abort/deadline, so both are ignored.
+fetch_read_coalescer::read_factory
+as_factory(fetch_read_coalescer::read_fn fn) {
+    return [fn = std::move(fn)](
+             ss::abort_source&, model::timeout_clock::time_point) mutable {
+        return std::move(fn);
+    };
+}
+
+fetch_read_coalescer::read_factory counting_read(
+  int& calls,
+  size_t bytes = 7,
+  model::offset hwm = model::offset{100},
+  model::offset lso = model::offset{100}) {
+    return as_factory([&calls, bytes, hwm, lso] {
+        ++calls;
+        return ss::make_ready_future<read_result>(make_result(bytes, hwm, lso));
+    });
+}
+
+fetch_read_coalescer::read_factory counting_error(int& calls) {
+    return as_factory([&calls] {
+        ++calls;
+        return ss::make_ready_future<read_result>(read_result(
+          error_code::offset_out_of_range,
+          model::offset{5},
+          model::offset{100},
+          model::offset{100}));
+    });
+}
+
+fetch_read_coalescer::read_factory counting_empty(int& calls) {
+    return as_factory([&calls] {
+        ++calls;
+        return ss::make_ready_future<read_result>(read_result(
+          model::offset{5}, model::offset{100}, model::offset{100}));
+    });
+}
+
+} // namespace
+
+TEST_CORO(FetchReadCoalescer, RetainAndShare) {
+    config::mock_property<bool> 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<bool> 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<shared_read> 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<bool> enabled{true};
+    fetch_read_coalescer c(test_cache_config(), enabled.bind());
+    auto key = make_key();
+
+    int calls = 0;
+    ss::promise<read_result> 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}, as_factory([&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<bool> 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<bool> 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<bool> 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<bool> 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<bool> 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}, as_factory([&calls] {
+              ++calls;
+              return ss::make_exception_future<read_result>(
+                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<bool> 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
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());
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",