From 035f28480e3f4368fa52fa69dbb3fc227e17cc50 Mon Sep 17 00:00:00 2001 From: Bartosz Piekny Date: Tue, 14 Jul 2026 12:39:59 +0200 Subject: [PATCH 1/2] kafka/client: recover from offset_out_of_range in consumer group fetch Pandaproxy's consumer group fetch started every fresh assignment at offset 0 and never advanced once retention moved the log start offset past it, so every fetch returned offset_out_of_range forever despite auto.offset.reset=earliest -- the only reset policy pandaproxy accepts. fetch_session now exposes three named operations instead of one overloaded apply(): apply() advances offsets from a delivered response, discard() advances only the session epoch, and reseed() sets a partition's offset directly. consumer::fetch() collects every broker's response, then: - reseeds out-of-range partitions to the broker-reported log_start_offset. earliest is exactly the log start, and the broker already returns it in the fetch response, so no separate ListOffsets is needed. - strips the out-of-range partitions from the response, since the pandaproxy serializer rejects any partition error. No offset advances past undelivered records, so nothing is silently skipped. - only on a dispatch failure -- a whole broker's response missing, where the topology may be stale -- discards the round and throws, so the retry in client::consumer_fetch() re-fetches and refreshes metadata. A round whose only outcome was the reseed carries no records, so instead of returning an empty poll and deferring the data -- which already exists at the reseeded offset -- to the client's next poll, fetch() repeats the round within the caller's timeout budget until it has data, nothing was reseeded, or the budget is spent. A round that delivered records (healthy partitions) returns immediately; an out-of-range sibling resumes on the next poll. Recovering per partition rather than discarding the whole round avoids re-reading the healthy partitions on every retention/trim edge. This mirrors the in-poll recovery of franz-go and the Java consumer (Confluent REST proxy): a timer-bounded poll() loop that resets the position and refetches across iterations, returning data once available or empty when the timer expires (apache/kafka 3.6): do { updateAssignmentMetadataIfNeeded(timer, false); // resets position final Fetch fetch = pollForFetches(timer); // (re)fetches if (!fetch.isEmpty()) { ...; return records; } } while (timer.notExpired()); return ConsumerRecords.empty(); poll loop: https://github.com/apache/kafka/blob/3.6/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L1174-L1207 out-of-range detect: https://github.com/apache/kafka/blob/3.6/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java#L654-L666 reset via ListOffsets: https://github.com/apache/kafka/blob/3.6/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java#L109-L116 Adds fetch_session unit tests for the split API. --- src/v/kafka/client/consumer.cc | 215 +++++++++++++++++++++-- src/v/kafka/client/consumer.h | 16 ++ src/v/kafka/client/fetch_session.cc | 42 ++++- src/v/kafka/client/fetch_session.h | 37 +++- src/v/kafka/client/test/fetch_session.cc | 148 +++++++++++++++- 5 files changed, 426 insertions(+), 32 deletions(-) diff --git a/src/v/kafka/client/consumer.cc b/src/v/kafka/client/consumer.cc index f52bfc0fe3f00..1978bb68132be 100644 --- a/src/v/kafka/client/consumer.cc +++ b/src/v/kafka/client/consumer.cc @@ -28,12 +28,19 @@ #include "ssx/future-util.h" #include +#include #include +#include +#include +#include #include #include #include +#include +#include #include +#include namespace kafka::client { @@ -78,6 +85,96 @@ 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> responses; +}; + +fetch_results collect_fetch_results( + const std::vector& req_brokers, + std::vector> 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). +/// +/// \return true if at least one partition was reseeded. +bool reseed_out_of_range( + absl::node_hash_map& fetch_sessions, + std::vector>& responses) { + bool reseeded = false; + 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); + reseeded = true; + } + } + } + return reseeded; +} + +/// \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 topics; + for (auto& topic : res.data.responses) { + chunked_vector 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); +} + +/// \brief Whether a fetch response carries any records to deliver. +/// +/// empty() is non-destructive, so this does not consume the batch reader the +/// caller later serializes. +bool has_records(fetch_response& res) { + for (auto& part : res) { + auto& record_set = part.partition_response->records; + if (record_set && !record_set->empty()) { + return true; + } + } + return false; +} + } // namespace detail consumer::consumer( @@ -425,14 +522,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 consumer::fetch( +consumer::broker_reqs_t consumer::build_fetch_requests( std::chrono::milliseconds timeout, std::optional 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 +571,107 @@ ss::future consumer::fetch( _config.fetch_max_bytes)}); } } + return broker_reqs; +} - 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); +ss::future> consumer::fetch_round( + std::chrono::milliseconds timeout, std::optional max_bytes) { + 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 req_brokers; + std::vector> 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 (pandaproxy only allows auto.offset.reset=earliest, + // i.e. the log start). They carry no records and are stripped below. + const bool reseeded = 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)); + } + + // Deliver the healthy partitions: 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); + } + + 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 std::pair{std::move(result), reseeded}; +} + +ss::future consumer::fetch( + std::chrono::milliseconds timeout, std::optional max_bytes) { + refresh_inactivity_timer(); + + // An out-of-range partition is reseeded to the log start and stripped from + // the response (the pandaproxy serializer rejects any partition error), so + // a round whose only outcome was the reseed carries no records. Rather than + // return that empty round -- forcing the client to poll again for data that + // already exists at the reseeded offset -- repeat the round within the + // caller's timeout budget, exactly what the client's next poll would do. + // This mirrors the timer-bounded do/while in the Java consumer's poll() + // (Confluent REST proxy): a round that delivered records returns at once, + // and if the budget is spent mid-recovery the empty round is returned and + // the reseeded partition resumes on the next poll. + // Seeded with an empty response for the degenerate case where the timeout + // budget is already spent and the loop body never runs. + const auto deadline = ss::lowres_clock::now() + timeout; + fetch_response result{ + .data = { + .throttle_time_ms{}, + .error_code = error_code::none, + .session_id = kafka::invalid_fetch_session_id}}; + bool reseeded = false; + while (ss::lowres_clock::now() < deadline) { + const auto remaining = std::max( + deadline - ss::lowres_clock::now(), + ss::lowres_clock::duration::zero()); + std::tie(result, reseeded) = co_await fetch_round( + std::chrono::duration_cast(remaining), + max_bytes); + + // Deliver as soon as there is data or when nothing was reseeded (a + // genuine empty long-poll). Otherwise repeat to resolve a round that + // reseeded but delivered nothing, until the timeout budget is spent. + if (detail::has_records(result) || !reseeded) { + break; + } + } + co_return std::move(result); } template diff --git a/src/v/kafka/client/consumer.h b/src/v/kafka/client/consumer.h index 11a87a4b00ef4..b032a25f47129 100644 --- a/src/v/kafka/client/consumer.h +++ b/src/v/kafka/client/consumer.h @@ -31,6 +31,7 @@ #include #include +#include namespace kafka::client { @@ -106,6 +107,21 @@ class consumer final : public ss::enable_lw_shared_from_this { ss::future 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 max_bytes); + + /// \brief Run one fetch round against every assigned broker: dispatch, + /// collect, reseed out-of-range partitions to the log start, advance each + /// session, and reduce into a single response with the reseeded (empty) + /// partitions stripped. Throws on a dispatch failure so the caller's retry + /// can refresh metadata. Returns the delivered response and whether any + /// partition was reseeded (i.e. the round may need repeating). + ss::future> fetch_round( + std::chrono::milliseconds timeout, std::optional max_bytes); + template requires requires(const RequestFactory v) { v.operator()(); } ss::future< diff --git a/src/v/kafka/client/fetch_session.cc b/src/v/kafka/client/fetch_session.cc index 6b2d2545c0160..c70f4567adfd5 100644 --- a/src/v/kafka/client/fetch_session.cc +++ b/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 diff --git a/src/v/kafka/client/fetch_session.h b/src/v/kafka/client/fetch_session.h index 147c4dd9af922..66edfdbe2a7af 100644 --- a/src/v/kafka/client/fetch_session.h +++ b/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 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< diff --git a/src/v/kafka/client/test/fetch_session.cc b/src/v/kafka/client/test/fetch_session.cc index ba063b08fa78a..1dfcd3aa21415 100644 --- a/src/v/kafka/client/test/fetch_session.cc +++ b/src/v/kafka/client/test/fetch_session.cc @@ -58,6 +58,67 @@ kafka::fetch_response make_fetch_response( return res; } +kafka::fetch_response make_out_of_range_fetch_response( + kafka::fetch_session_id s_id, + model::topic_partition_view tpv, + model::offset log_start_offset) { + kafka::fetch_response res{ + .data = { + .throttle_time_ms = std::chrono::milliseconds{0}, + .error_code = kafka::error_code::none, + .session_id = s_id, + .responses{}}}; + kafka::fetch_response::partition p{.topic = tpv.topic}; + p.partitions.push_back( + kafka::fetch_response::partition_response{ + .partition_index = tpv.partition, + .error_code = kafka::error_code::offset_out_of_range, + .high_watermark = model::offset{-1}, + .last_stable_offset = model::offset{-1}, + .log_start_offset = log_start_offset, + .aborted_transactions = {}, + .records{std::nullopt}}); + res.data.responses.push_back(std::move(p)); + return res; +} + +// A response for two partitions of one topic: `ok_tpv` succeeds with +// `ok_record_set`, `oor_partition` fails with offset_out_of_range. +kafka::fetch_response make_partial_out_of_range_fetch_response( + kafka::fetch_session_id s_id, + model::topic_partition_view ok_tpv, + std::optional ok_record_set, + model::partition_id oor_partition, + model::offset log_start_offset) { + kafka::fetch_response res{ + .data = { + .throttle_time_ms = std::chrono::milliseconds{0}, + .error_code = kafka::error_code::none, + .session_id = s_id, + .responses{}}}; + kafka::fetch_response::partition p{.topic = ok_tpv.topic}; + p.partitions.push_back( + kafka::fetch_response::partition_response{ + .partition_index = ok_tpv.partition, + .error_code = kafka::error_code::none, + .high_watermark = model::offset{-1}, + .last_stable_offset = model::offset{-1}, + .log_start_offset = model::offset{-1}, + .aborted_transactions = {}, + .records{std::move(ok_record_set)}}); + p.partitions.push_back( + kafka::fetch_response::partition_response{ + .partition_index = oor_partition, + .error_code = kafka::error_code::offset_out_of_range, + .high_watermark = model::offset{-1}, + .last_stable_offset = model::offset{-1}, + .log_start_offset = log_start_offset, + .aborted_transactions = {}, + .records{std::nullopt}}); + res.data.responses.push_back(std::move(p)); + return res; +} + struct context { const kafka::fetch_session_id fetch_session_id{42}; const model::topic_partition tp{ @@ -67,13 +128,13 @@ struct context { kafka::initial_fetch_session_epoch}; model::offset expected_offset{0}; - bool + void apply_fetch_response(kc::fetch_session& s, std::optional count) { auto res = make_fetch_response( fetch_session_id, tp, make_record_set(expected_offset, count)); expected_offset += count.value_or(0); ++expected_epoch; - return s.apply(res); + s.apply(res); } }; @@ -86,13 +147,13 @@ SEASTAR_THREAD_TEST_CASE(test_fetch_session) { BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{0}); // Apply some records - BOOST_REQUIRE(ctx.apply_fetch_response(s, 8)); + ctx.apply_fetch_response(s, 8); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); // Apply more records - BOOST_REQUIRE(ctx.apply_fetch_response(s, 8)); + ctx.apply_fetch_response(s, 8); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); @@ -103,13 +164,13 @@ SEASTAR_THREAD_TEST_CASE(test_fetch_session_null_record_set) { kc::fetch_session s; // Apply some records - BOOST_REQUIRE(ctx.apply_fetch_response(s, 8)); + ctx.apply_fetch_response(s, 8); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); // Apply nullopt record_set - BOOST_REQUIRE(ctx.apply_fetch_response(s, std::nullopt)); + ctx.apply_fetch_response(s, std::nullopt); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); @@ -120,18 +181,87 @@ SEASTAR_THREAD_TEST_CASE(test_fetch_session_empty_record_set) { kc::fetch_session s; // Apply some records - BOOST_REQUIRE(ctx.apply_fetch_response(s, 8)); + ctx.apply_fetch_response(s, 8); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); // Apply 0 records - BOOST_REQUIRE(ctx.apply_fetch_response(s, 0)); + ctx.apply_fetch_response(s, 0); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); } +// CORE-16844: reseed() lets the caller correct the tracked offset after +// offset_out_of_range, so a fetch doesn't repeat the same stale offset. +SEASTAR_THREAD_TEST_CASE(test_fetch_session_reseed) { + context ctx; + kc::fetch_session s; + + BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{0}); + + s.reseed(ctx.tp, model::offset{48}); + + BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{48}); +} + +// CORE-16844: apply()/discard() do not reseed offset_out_of_range +// partitions themselves -- that's the caller's job via reseed(), done once +// while scanning every broker's response to decide whether to retry (see +// consumer::fetch()). +SEASTAR_THREAD_TEST_CASE(test_fetch_session_apply_does_not_reseed) { + context ctx; + kc::fetch_session s; + + auto res = make_out_of_range_fetch_response( + ctx.fetch_session_id, ctx.tp, model::offset{48}); + s.apply(res); + + BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{0}); +} + +// CORE-16844: when any partition on any broker returns offset_out_of_range, +// consumer::fetch() discards the combined response and retries, calling +// discard() so that records fetched but never delivered are not +// skipped by the retry — that would be silent data loss. The epoch must +// still advance, to stay in step with the broker-side session state. +SEASTAR_THREAD_TEST_CASE( + test_fetch_session_no_offset_advance_when_response_discarded) { + context ctx; + kc::fetch_session s; + + const model::topic_partition oor_tp{ctx.tp.topic, model::partition_id{3}}; + + auto discarded = make_partial_out_of_range_fetch_response( + ctx.fetch_session_id, + ctx.tp, + make_record_set(model::offset{0}, 8), + oor_tp.partition, + model::offset{48}); + s.discard(discarded); + + // The healthy partition's offset stays put; its records are re-fetched + // by the retry. + BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{0}); + // discard() does not reseed on its own. + BOOST_REQUIRE_EQUAL(s.offset(oor_tp), model::offset{0}); + // The epoch advances regardless: the broker processed the request. + BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch + 1); + + // The retry delivers the response to the client, so offsets advance. + auto delivered = make_partial_out_of_range_fetch_response( + ctx.fetch_session_id, + ctx.tp, + make_record_set(model::offset{0}, 8), + oor_tp.partition, + model::offset{48}); + s.apply(delivered); + + BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{8}); + BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch + 2); +} + SEASTAR_THREAD_TEST_CASE(test_fetch_session_make_offset_commit_request_all) { context ctx; kc::fetch_session s; @@ -141,7 +271,7 @@ SEASTAR_THREAD_TEST_CASE(test_fetch_session_make_offset_commit_request_all) { BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), model::offset{0}); // Apply some records - BOOST_REQUIRE(ctx.apply_fetch_response(s, 8)); + ctx.apply_fetch_response(s, 8); BOOST_REQUIRE_EQUAL(s.id(), ctx.fetch_session_id); BOOST_REQUIRE_EQUAL(s.epoch(), ctx.expected_epoch); BOOST_REQUIRE_EQUAL(s.offset(ctx.tp), ctx.expected_offset); From f35db722acfe8b2a7d5ad4e146aae61d00c11d81 Mon Sep 17 00:00:00 2001 From: Bartosz Piekny Date: Tue, 14 Jul 2026 12:39:59 +0200 Subject: [PATCH 2/2] tests/pandaproxy: cover offset_out_of_range recovery end-to-end Trim a topic's log prefix past offset 0 and assert a fresh consumer group polls its way to the records at the new log start offset, producing one record per call so each lands in its own batch and the trim offset falls on a batch boundary, as real retention does. A second test trims only one of two partitions and asserts the healthy sibling's records are delivered while the out-of-range partition recovers, covering the per-partition recovery path. --- tests/rptest/tests/pandaproxy_test.py | 247 ++++++++++++++++++++++++++ 1 file changed, 247 insertions(+) diff --git a/tests/rptest/tests/pandaproxy_test.py b/tests/rptest/tests/pandaproxy_test.py index 7bf9e602b7583..074d71de9b0f3 100644 --- a/tests/rptest/tests/pandaproxy_test.py +++ b/tests/rptest/tests/pandaproxy_test.py @@ -2316,6 +2316,253 @@ def do_coordinator_change(utils: GroupCoordinatorTransferUtils): self.logger.debug("Test offset commit: after second coordinator change") do_consumer_offset_commit(new_offset=1) + @cluster(num_nodes=3) + def test_consumer_group_fetch_after_prefix_trim(self): + """ + CORE-16844: a fresh consumer group must recover once retention + moves the log start offset past 0, not fetch offset 0 forever. + """ + num_records = 20 + trim_offset = 10 + + self.logger.info(f"Producing {num_records} records to topic: {self.topic}") + # One produce call per record, so each lands in its own batch: real + # retention deletes whole segments, which are batch-aligned, so the + # new log start offset always lands on a batch boundary too. A + # single multi-record produce call would put everything in one + # batch, and trimming into the middle of it is not something + # retention does. + for i in range(num_records): + produce_result_raw = self._produce_topic( + self.topic, + json.dumps({"records": [{"value": {"i": i}, "partition": 0}]}), + headers=HTTP_PRODUCE_JSON_V2_TOPIC_HEADERS, + ) + assert produce_result_raw.status_code == requests.codes.ok + + self.logger.info(f"Trimming {self.topic}/0 prefix to offset {trim_offset}") + rpk = RpkTool(self.redpanda) + trim_result = rpk.trim_prefix(self.topic, offset=trim_offset, partitions=[0]) + assert len(trim_result) == 1 + assert trim_result[0].new_start_offset == trim_offset, trim_result[0] + + self.logger.info("Create a consumer group with auto.offset.reset=earliest") + group_id = f"pandaproxy-group-{uuid.uuid4()}" + cc_res = self._create_consumer(group_id) + assert cc_res.status_code == requests.codes.ok + c0 = Consumer(cc_res.json(), self.logger) + + sc_res = c0.subscribe([self.topic]) + assert sc_res.status_code == requests.codes.no_content + + self.logger.info( + "Fetch should recover from offset_out_of_range and return " + "records starting at the new log start offset" + ) + # The out-of-range partition is reseeded to the new log start and the + # fetch recovers in-poll, so records arrive from the new log start + # offset. The initial group join may still yield an empty poll, so + # poll until they arrive. + fetch_result: list = [] + + def fetched_records(): + cf_res = c0.fetch(headers=HTTP_CONSUMER_FETCH_JSON_V2_HEADERS) + assert cf_res.status_code == requests.codes.ok, ( + f"Expected 200, got {cf_res.status_code}: {cf_res.text}" + ) + fetch_result.extend(cf_res.json()) + return len(fetch_result) > 0 + + wait_until( + fetched_records, + timeout_sec=30, + backoff_sec=0, + err_msg="consumer never recovered after prefix trim", + ) + assert fetch_result[0]["offset"] == trim_offset, ( + f"Expected first fetched record at offset {trim_offset}, " + f"got {fetch_result[0]['offset']}" + ) + + @cluster(num_nodes=3) + def test_consumer_group_fetch_no_data_loss_on_sibling_out_of_range(self): + """ + CORE-16844: when one partition in a fetch round returns + offset_out_of_range, the whole round is discarded and retried so its + offset can be reseeded. A healthy sibling partition fetched in the + same round must NOT have its offset advanced by that discarded round: + its records were never delivered, and advancing past them would make + the retry skip them -- silent data loss. This exercises that decision + in consumer::fetch() end-to-end, across a partition that is trimmed + and one that is not. + """ + topic = f"cg-sibling-{uuid.uuid4()}" + healthy_partition = 0 + trimmed_partition = 1 + num_records = 20 + trim_offset = 10 + + self.logger.info(f"Creating 2-partition topic: {topic}") + rpk = RpkTool(self.redpanda) + rpk.create_topic(topic, partitions=2) + + # One produce call per record, so each lands in its own batch and the + # trim offset falls on a batch boundary, matching what real retention + # (which only deletes whole, batch-aligned segments) produces. + self.logger.info(f"Producing {num_records} records to each partition") + for p in (healthy_partition, trimmed_partition): + for i in range(num_records): + produce_result_raw = self._produce_topic( + topic, + json.dumps({"records": [{"value": {"i": i}, "partition": p}]}), + headers=HTTP_PRODUCE_JSON_V2_TOPIC_HEADERS, + ) + assert produce_result_raw.status_code == requests.codes.ok + + self.logger.info( + f"Trimming only {topic}/{trimmed_partition} to offset {trim_offset}" + ) + trim_result = rpk.trim_prefix( + topic, offset=trim_offset, partitions=[trimmed_partition] + ) + assert len(trim_result) == 1 + assert trim_result[0].new_start_offset == trim_offset, trim_result[0] + + self.logger.info("Create a consumer group with auto.offset.reset=earliest") + group_id = f"pandaproxy-group-{uuid.uuid4()}" + cc_res = self._create_consumer(group_id) + assert cc_res.status_code == requests.codes.ok + c0 = Consumer(cc_res.json(), self.logger) + + sc_res = c0.subscribe([topic]) + assert sc_res.status_code == requests.codes.no_content + + # The healthy partition delivers offsets 0..num_records-1, the trimmed + # one delivers trim_offset..num_records-1. + expected = { + healthy_partition: set(range(num_records)), + trimmed_partition: set(range(trim_offset, num_records)), + } + expected_total = sum(len(v) for v in expected.values()) + + # Accumulate across fetches: a single fetch may not drain both + # partitions at once, and the first round is discarded and retried + # internally to recover from offset_out_of_range. + seen: dict[int, set[int]] = { + healthy_partition: set(), + trimmed_partition: set(), + } + + def fetched_everything(): + cf_res = c0.fetch(headers=HTTP_CONSUMER_FETCH_JSON_V2_HEADERS) + assert cf_res.status_code == requests.codes.ok, ( + f"Expected 200, got {cf_res.status_code}: {cf_res.text}" + ) + for r in cf_res.json(): + seen[r["partition"]].add(r["offset"]) + return sum(len(v) for v in seen.values()) >= expected_total + + wait_until( + fetched_everything, + timeout_sec=30, + backoff_sec=0, + err_msg="Timed out before fetching all records from both partitions", + ) + + # The healthy sibling's records must all survive: if the discarded + # round had advanced its offset, offsets 0..trim_offset-1 would be + # missing here even though the fetch returned HTTP 200. + assert seen == expected, f"Expected {expected}, got {seen}" + + @cluster(num_nodes=3) + def test_consumer_group_fetch_recovers_out_of_range_in_single_poll(self): + """ + CORE-16844: once a consumer's fetch position is established, a fetch + that hits offset_out_of_range must recover and return the reseeded + records in the SAME poll, not return an empty poll and defer the data + to the next one. This matches the in-poll recovery of franz-go and the + Java (Confluent REST) consumer, both of which block the poll through + the internal reset+refetch. The consumer is drained first so the group + join -- which can itself yield an empty poll -- cannot be mistaken for + the out-of-range recovery under test. + """ + num_records = 20 + trim_offset = 30 + + # One produce call per record, so each lands in its own batch and the + # trim offset falls on a batch boundary (see the sibling test). + def produce(lo, hi): + for i in range(lo, hi): + produce_result_raw = self._produce_topic( + self.topic, + json.dumps({"records": [{"value": {"i": i}, "partition": 0}]}), + headers=HTTP_PRODUCE_JSON_V2_TOPIC_HEADERS, + ) + assert produce_result_raw.status_code == requests.codes.ok + + self.logger.info(f"Producing {num_records} records to topic: {self.topic}") + produce(0, num_records) + + self.logger.info("Create a consumer group with auto.offset.reset=earliest") + group_id = f"pandaproxy-group-{uuid.uuid4()}" + cc_res = self._create_consumer(group_id) + assert cc_res.status_code == requests.codes.ok + c0 = Consumer(cc_res.json(), self.logger) + + sc_res = c0.subscribe([self.topic]) + assert sc_res.status_code == requests.codes.no_content + + # Drain the initial records: this completes the group join and advances + # the fetch position to num_records, so the only obstacle to the fetch + # below is the out-of-range condition we create next. + self.logger.info("Draining initial records to settle the group join") + drained: list = [] + + def drained_all(): + cf_res = c0.fetch(headers=HTTP_CONSUMER_FETCH_JSON_V2_HEADERS) + assert cf_res.status_code == requests.codes.ok, ( + f"Expected 200, got {cf_res.status_code}: {cf_res.text}" + ) + drained.extend(cf_res.json()) + return len(drained) >= num_records + + wait_until( + drained_all, + timeout_sec=30, + backoff_sec=0, + err_msg="consumer never drained the initial records", + ) + + # Produce more, then trim past the settled fetch position so the next + # fetch asks for an offset below the new log start -> offset_out_of_range. + self.logger.info(f"Producing records {num_records}..{2 * num_records - 1}") + produce(num_records, 2 * num_records) + + self.logger.info(f"Trimming {self.topic}/0 prefix to offset {trim_offset}") + rpk = RpkTool(self.redpanda) + trim_result = rpk.trim_prefix(self.topic, offset=trim_offset, partitions=[0]) + assert len(trim_result) == 1 + assert trim_result[0].new_start_offset == trim_offset, trim_result[0] + + # A SINGLE fetch must recover and return records from the new log start + # offset. Before the in-poll fix this first fetch returned empty (the + # out-of-range partition was reseeded and stripped) and the data only + # arrived on a later poll. + self.logger.info("A single fetch must return the reseeded records") + cf_res = c0.fetch(headers=HTTP_CONSUMER_FETCH_JSON_V2_HEADERS) + assert cf_res.status_code == requests.codes.ok, ( + f"Expected 200, got {cf_res.status_code}: {cf_res.text}" + ) + recovered = cf_res.json() + assert len(recovered) > 0, ( + "expected the out-of-range fetch to recover in a single poll, " + "got an empty response" + ) + assert recovered[0]["offset"] == trim_offset, ( + f"expected first recovered record at offset {trim_offset}, " + f"got {recovered[0]['offset']}" + ) + class PandaProxyCompressedBatchesTest(PandaProxyEndpoints): topics = [