[CORE-16844] kafka/client: recover from offset_out_of_range in consumer group fetch

src/v/kafka/client/consumer.cc

@@ -29,11 +29,16 @@
 
 #include <seastar/core/gate.hh>
 #include <seastar/core/sleep.hh>
+#include <seastar/core/when_all.hh>
+#include <seastar/coroutine/exception.hh>
 
+#include <algorithm>
 #include <chrono>
 #include <exception>
 #include <iterator>
+#include <optional>
 #include <utility>
+#include <vector>
 
 namespace kafka::client {
 
@@ -78,6 +83,77 @@ reduce_fetch_response(fetch_response result, fetch_response val) {
     return result;
 };
 
+/// \brief The outcome of dispatching a fetch to every assigned broker: the
+/// first dispatch-level failure (if any), and the response of every broker
+/// that did not fail.
+struct fetch_results {
+    std::exception_ptr dispatch_failure;
+    std::vector<std::pair<shared_broker_t, fetch_response>> responses;
+};
+
+fetch_results collect_fetch_results(
+  const std::vector<shared_broker_t>& req_brokers,
+  std::vector<ss::future<fetch_response>> results) {
+    fetch_results out;
+    out.responses.reserve(results.size());
+    for (size_t i = 0; i < results.size(); ++i) {
+        if (results[i].failed()) {
+            auto ex = results[i].get_exception();
+            if (!out.dispatch_failure) {
+                out.dispatch_failure = std::move(ex);
+            }
+            continue;
+        }
+        out.responses.emplace_back(req_brokers[i], results[i].get());
+    }
+    return out;
+}
+
+/// \brief Reseed every offset_out_of_range partition across every broker's
+/// response to the log_start_offset the broker reported (pandaproxy only
+/// allows auto.offset.reset=earliest, which is exactly the log start).
+void reseed_out_of_range(
+  absl::node_hash_map<shared_broker_t, fetch_session>& fetch_sessions,
+  std::vector<std::pair<shared_broker_t, fetch_response>>& responses) {
+    for (auto& [broker, res] : responses) {
+        for (auto& part : res) {
+            if (
+              part.partition_response->error_code
+              == error_code::offset_out_of_range) {
+                model::topic_partition tp{
+                  part.partition->topic,
+                  part.partition_response->partition_index};
+                fetch_sessions[broker].reseed(
+                  tp, part.partition_response->log_start_offset);
+            }
+        }
+    }
+}
+
+/// \brief Drop offset_out_of_range partitions from a fetch response.
+///
+/// They were reseeded (see reseed_out_of_range) and carry no records, so the
+/// caller delivers the remaining healthy partitions now and the reseeded ones
+/// resume from the corrected offset on the client's next fetch. Stripping them
+/// also keeps the pandaproxy serializer -- which rejects any partition error --
+/// from turning a recoverable round into an HTTP error.
+void strip_out_of_range(fetch_response& res) {
+    chunked_vector<fetch_response::partition> topics;
+    for (auto& topic : res.data.responses) {
+        chunked_vector<fetch_response::partition_response> partitions;
+        for (auto& part : topic.partitions) {
+            if (part.error_code != error_code::offset_out_of_range) {
+                partitions.push_back(std::move(part));
+            }
+        }
+        if (!partitions.empty()) {
+            topic.partitions = std::move(partitions);
+            topics.push_back(std::move(topic));
+        }
+    }
+    res.data.responses = std::move(topics);
+}
+
 } // namespace detail
 
 consumer::consumer(
@@ -425,14 +501,14 @@ consumer::dispatch_fetch(broker_reqs_t::value_type br) {
         throw broker_error(broker->id(), res.data.error_code);
     }
 
-    _fetch_sessions[broker].apply(res);
+    // Session state is deliberately not applied here: whether the tracked
+    // offsets may advance depends on the responses of ALL brokers, which
+    // only consumer::fetch() has in hand.
     co_return res;
 }
 
-ss::future<fetch_response> consumer::fetch(
+consumer::broker_reqs_t consumer::build_fetch_requests(
   std::chrono::milliseconds timeout, std::optional<int32_t> max_bytes) {
-    refresh_inactivity_timer();
-    // Split requests by broker
     broker_reqs_t broker_reqs;
     for (const auto& [t, ps] : _assignment) {
         for (const auto& p : ps) {
@@ -474,17 +550,68 @@ ss::future<fetch_response> consumer::fetch(
                   _config.fetch_max_bytes)});
         }
     }
+    return broker_reqs;
+}
+
+ss::future<fetch_response> consumer::fetch(
+  std::chrono::milliseconds timeout, std::optional<int32_t> max_bytes) {
+    refresh_inactivity_timer();
+    auto broker_reqs = build_fetch_requests(timeout, max_bytes);
+
+    // Dispatch to all brokers concurrently, then wait for every result
+    // before touching any session state.
+    std::vector<shared_broker_t> req_brokers;
+    std::vector<ss::future<fetch_response>> dispatched;
+    req_brokers.reserve(broker_reqs.size());
+    dispatched.reserve(broker_reqs.size());
+    for (auto& br : broker_reqs) {
+        req_brokers.push_back(br.first);
+        dispatched.push_back(dispatch_fetch(std::move(br)));
+    }
+    auto results = co_await ss::when_all(dispatched.begin(), dispatched.end());
+
+    auto [dispatch_failure, responses] = detail::collect_fetch_results(
+      req_brokers, std::move(results));
+
+    // Out-of-range partitions are reseeded to the broker-reported
+    // log_start_offset. They carry no records and are stripped from the
+    // delivered response below, so the client resumes them from the
+    // corrected offset on its next fetch (pandaproxy only allows
+    // auto.offset.reset=earliest, i.e. the log start).
+    detail::reseed_out_of_range(_fetch_sessions, responses);
+
+    // A dispatch failure means a whole broker's response never arrived and
+    // the topology may be stale, so discard the round and throw: the retry
+    // in client::consumer_fetch() re-fetches and refreshes metadata. No
+    // offset may advance here -- those records were never delivered, and
+    // advancing would make the retry skip them (silent data loss). discard()
+    // still advances each session's epoch to stay in step with the broker.
+    if (dispatch_failure) {
+        for (auto& [broker, res] : responses) {
+            _fetch_sessions[broker].discard(res);
+        }
+        co_await ss::coroutine::return_exception_ptr(
+          std::move(dispatch_failure));
+    }
+
+    // No dispatch failure: deliver the healthy partitions now. apply()
+    // advances the offsets of partitions that returned records and skips the
+    // out-of-range ones, which were reseeded above and are stripped below.
+    for (auto& [broker, res] : responses) {
+        _fetch_sessions[broker].apply(res);
+    }
 
-    co_return co_await ss::map_reduce(
-      std::make_move_iterator(broker_reqs.begin()),
-      std::make_move_iterator(broker_reqs.end()),
-      [this](broker_reqs_t::value_type br) {
-          return dispatch_fetch(std::move(br));
-      },
-      fetch_response{
-        .data
-        = {.throttle_time_ms{}, .error_code = error_code::none, .session_id = kafka::invalid_fetch_session_id}},
-      detail::reduce_fetch_response);
+    fetch_response result{
+      .data = {
+        .throttle_time_ms{},
+        .error_code = error_code::none,
+        .session_id = kafka::invalid_fetch_session_id}};
+    for (auto& [broker, res] : responses) {
+        result = detail::reduce_fetch_response(
+          std::move(result), std::move(res));
+    }
+    detail::strip_out_of_range(result);
+    co_return result;
 }
 
 template<typename request_factory>

src/v/kafka/client/consumer.h

@@ -106,6 +106,12 @@ class consumer final : public ss::enable_lw_shared_from_this<consumer> {
 
     ss::future<fetch_response> dispatch_fetch(broker_reqs_t::value_type br);
 
+    /// \brief Build one fetch_request per broker leading a partition in the
+    /// current assignment, seeded with each broker's fetch_session id/epoch
+    /// and each partition's tracked fetch offset.
+    broker_reqs_t build_fetch_requests(
+      std::chrono::milliseconds timeout, std::optional<int32_t> max_bytes);
+
     template<typename RequestFactory>
     requires requires(const RequestFactory v) { v.operator()(); }
     ss::future<

src/v/kafka/client/fetch_session.cc

@@ -26,30 +26,56 @@ model::offset fetch_session::offset(model::topic_partition_view tpv) const {
     return part_it->second;
 }
 
-bool fetch_session::apply(fetch_response& res) {
+void fetch_session::reseed(
+  model::topic_partition_view tpv, model::offset new_offset) {
+    _offsets[model::topic{tpv.topic}][tpv.partition] = new_offset;
+}
+
+void fetch_session::update_session_state(const fetch_response& res) {
     if (_id == invalid_fetch_session_id) {
         _id = fetch_session_id{res.data.session_id};
     }
     vassert(res.data.session_id == _id, "session mismatch: {}", *this);
-
     ++_epoch;
+}
+
+void fetch_session::rehash_offsets() {
+    for (auto& topic : _offsets) {
+        topic.second.rehash(topic.second.size());
+    }
+}
+
+void fetch_session::apply(fetch_response& res) {
+    update_session_state(res);
+
     for (auto& part : res) {
+        // offset_out_of_range partitions are reseeded by the caller (which
+        // already scans every partition to build the retry decision) before
+        // apply() is called, so their non-none error_code just skips them
+        // here without touching the reseeded offset.
         if (part.partition_response->error_code != error_code::none) {
             continue;
         }
-        const auto& topic = part.partition->topic;
-        const auto p_id = part.partition_response->partition_index;
         auto& record_set = part.partition_response->records;
         if (!record_set || record_set->empty()) {
             continue;
         }
 
+        const auto& topic = part.partition->topic;
+        const auto p_id = part.partition_response->partition_index;
         _offsets[topic][p_id] = ++record_set->last_offset();
     }
-    for (auto& topic : _offsets) {
-        topic.second.rehash(topic.second.size());
-    }
-    return true;
+    rehash_offsets();
+}
+
+void fetch_session::discard(fetch_response& res) {
+    // The records in this response were never delivered to the caller, so
+    // no partition's offset may advance here: that would make the retry
+    // skip them, silently losing data. Only the session bookkeeping happens
+    // -- but reseed() (called by the caller before discard()) may
+    // still have grown _offsets, so compact it here too.
+    update_session_state(res);
+    rehash_offsets();
 }
 
 std::vector<offset_commit_request_topic>

src/v/kafka/client/fetch_session.h

@@ -39,13 +39,48 @@ class fetch_session {
     void id(kafka::fetch_session_id id) { _id = id; }
     kafka::fetch_session_epoch epoch() const { return _epoch; }
     model::offset offset(model::topic_partition_view tpv) const;
-    bool apply(fetch_response& res);
+
+    /// \brief Directly set the tracked fetch offset for a partition, e.g. to
+    /// recover from offset_out_of_range by seeking to the log_start_offset
+    /// the broker reported (pandaproxy only allows
+    /// auto.offset.reset=earliest).
+    void reseed(model::topic_partition_view tpv, model::offset new_offset);
+
+    /// \brief Update session state from a fetch response delivered to the
+    /// caller.
+    ///
+    /// The session epoch advances: the broker has processed the request and
+    /// advanced its side of the session. Offsets of partitions that returned
+    /// records advance too.
+    void apply(fetch_response& res);
+
+    /// \brief Update session state from a fetch response that is being
+    /// discarded and re-fetched, e.g. because a sibling broker's fetch in the
+    /// same round failed.
+    ///
+    /// The session epoch still advances, to stay in step with the
+    /// broker-side session state, but offsets of partitions that returned
+    /// records do NOT advance: those records were never delivered to the
+    /// caller, and advancing past them would make the retry skip them,
+    /// silently losing data.
+    void discard(fetch_response& res);
+
     std::vector<kafka::offset_commit_request_topic>
     make_offset_commit_request() const;
 
     fmt::iterator format_to(fmt::iterator it) const;
 
 private:
+    /// \brief Common session bookkeeping shared by apply() and
+    /// discard(): seed/validate the session id and advance the
+    /// epoch, since the broker has processed the request regardless of what
+    /// happens to the response on our side.
+    void update_session_state(const fetch_response& res);
+
+    /// \brief Compact each partition-offset map to its current size, e.g.
+    /// after reseed() or apply() may have grown it.
+    void rehash_offsets();
+
     kafka::fetch_session_id _id{kafka::invalid_fetch_session_id};
     kafka::fetch_session_epoch _epoch{kafka::initial_fetch_session_epoch};
     absl::node_hash_map<