[v25.2.x] [CORE-16844] kafka/client: recover from offset_out_of_range in consumer group fetch - #31206
Merged
bartoszpiekny-redpanda merged 3 commits intoJul 21, 2026
Conversation
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<K, V> 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.
(cherry picked from commit f12f363)
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. (cherry picked from commit 80417ce)
The backported fetch_session.cc was formatted by dev's clang-format (LLVM 22); v25.2.x lints with clang-format 18, which indents the designated initializers differently. Reformat to satisfy the release branch's toolchain.
Collaborator
Author
bartoszpiekny-redpanda
approved these changes
Jul 21, 2026
bartoszpiekny-redpanda
merged commit Jul 21, 2026
2fe0fc0
into
redpanda-data:v25.2.x
17 checks passed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Backport of PR #31064
Conflict details
<seastar/core/*>include block — the target branch already had#include <seastar/core/loop.hh>while the cherry-picked commit added#include <seastar/core/lowres_clock.hh>. Both includes are needed (loop.hh on the target branch, lowres_clock.hh for the new fetch-round timeout logic), so both were kept in sorted order.