[CORE-16844] kafka/client: recover from offset_out_of_range in consumer group fetch#31064
Draft
bartoszpiekny-redpanda wants to merge 1 commit into
Conversation
bartoszpiekny-redpanda
force-pushed
the
CORE-16844-fix-offset_out_of_range-after-retention
branch
2 times, most recently
from
9ca82a8 to
2293be2
Compare
bartoszpiekny-redpanda
force-pushed
the
CORE-16844-fix-offset_out_of_range-after-retention
branch
from
2293be2 to
940d6dd
Compare
Pandaproxy's consumer group fetch always started a fresh partition assignment at offset 0 and never advanced once retention moved the log start offset past it, regardless of auto.offset.reset=earliest requested at consumer creation. Every fetch after that point returned offset_out_of_range forever, since fetch_session::apply() silently discarded the error without correcting its tracked offset. fetch_session::apply() now seeds the tracked offset from the log_start_offset reported alongside offset_out_of_range, since pandaproxy only ever allows the earliest reset policy. dispatch_fetch() throws once the offset has been corrected, so the existing gated_retry_with_mitigation retry in client::consumer_fetch() re-fetches with the corrected offset before returning to the REST client, which has no way to control the fetch offset itself. Adds a fetch_session unit test covering the offset_out_of_range recovery, and a pandaproxy ducktape regression test that trims a topic's log prefix and asserts a fresh consumer group fetch still succeeds.
bartoszpiekny-redpanda
force-pushed
the
CORE-16844-fix-offset_out_of_range-after-retention
branch
from
940d6dd to
eb5dd14
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Fixes Pandaproxy consumer-group fetch getting stuck returning offset_out_of_range after log retention/prefix-truncation advances a partition’s log start offset beyond the consumer’s initial fetch offset.
Changes:
- Update
fetch_session::apply()to seed the tracked fetch offset fromlog_start_offsetwhen a partition returnsoffset_out_of_range. - Trigger a retry from the Kafka client consumer fetch path so Pandaproxy re-fetches using the corrected offset.
- Add unit and ducktape regression tests covering prefix-trim recovery.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| tests/rptest/tests/pandaproxy_test.py | Adds a ducktape regression test that trims a topic prefix and verifies a fresh consumer-group fetch still succeeds. |
| src/v/kafka/client/test/fetch_session.cc | Adds a unit test ensuring fetch_session::apply() updates tracked offsets on offset_out_of_range. |
| src/v/kafka/client/fetch_session.cc | Implements offset correction on offset_out_of_range using log_start_offset. |
| src/v/kafka/client/consumer.cc | Throws on offset_out_of_range after applying the response to force a retry before returning to Pandaproxy. |
Comment on lines
429
to
+444
| _fetch_sessions[broker].apply(res); | ||
|
|
||
| // Offset just corrected above; retry so the pandaproxy client, which | ||
| // can't control the fetch offset itself, never sees this error. | ||
| // Caught and retried by gated_retry_with_mitigation() in | ||
| // client::consumer_fetch. | ||
| auto it = std::find_if(res.begin(), res.end(), [](auto& part) { | ||
| return part.partition_response->error_code | ||
| == error_code::offset_out_of_range; | ||
| }); | ||
| if (it != res.end()) { | ||
| throw partition_error( | ||
| model::topic_partition{ | ||
| it->partition->topic, it->partition_response->partition_index}, | ||
| error_code::offset_out_of_range); | ||
| } |
Collaborator
Retry command for Build#86958please wait until all jobs are finished before running the slash command |
Collaborator
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
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.
Pandaproxy's consumer group fetch always started a fresh partition assignment at offset 0 and never advanced once retention moved the log start offset past it, regardless of auto.offset.reset=earliest requested at consumer creation. Every fetch after that point returned offset_out_of_range forever, since fetch_session::apply() silently discarded the error without correcting its tracked offset.
fetch_session::apply() now seeds the tracked offset from the log_start_offset reported alongside offset_out_of_range, since pandaproxy only ever allows the earliest reset policy. dispatch_fetch() throws once the offset has been corrected, so the existing gated_retry_with_mitigation retry in client::consumer_fetch() re-fetches with the corrected offset before returning to the REST client, which has no way to control the fetch offset itself.
Adds a fetch_session unit test covering the offset_out_of_range recovery, and a pandaproxy ducktape regression test that trims a topic's log prefix and asserts a fresh consumer group fetch still succeeds.
Fixes:
Backports Required
Release Notes