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

[bartoszpiekny-redpanda]

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, because the client tracked its own fetch offset and had no path to correct it — and the REST client cannot control the fetch offset itself.

fetch_session is split into three named operations: apply() advances the tracked offset from a delivered response, discard() advances only the session epoch, and reseed() sets an offset directly. On offset_out_of_range, consumer::fetch() reseeds the affected partitions to the log_start_offset the broker reports alongside the error — earliest is exactly the log start, and the broker already returns it in the fetch response, so no separate ListOffsets is needed since pandaproxy only ever allows the earliest reset policy. Healthy partitions are delivered immediately (their offsets advance) and the out-of-range ones are stripped from the response (the serializer rejects any partition error). A round whose only outcome was the reseed carries no records, so rather than return an empty poll and defer 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, so a single fetch returns the recovered records. Only a dispatch failure, where a whole broker's response is missing and the topology may be stale, discards the round and throws so the existing gated_retry_with_mitigation in client::consumer_fetch() re-fetches and refreshes metadata.

Adds fetch_session unit tests covering the split apply/discard/reseed API, and pandaproxy ducktape regression tests: one trims a topic's log prefix and asserts a fresh consumer group polls its way to the records at the new log start offset; one trims a single partition of a two-partition topic and asserts the healthy sibling's records are delivered while the out-of-range partition recovers; and one drains a settled consumer, then trims past its position and asserts a single fetch returns the reseeded records in the same poll.

Design notes

1. auto.offset.reset is earliest-only and is not plumbed into the client

The HTTP Proxy accepts a single reset policy. The consumer-create handler rejects anything else with a 400 before a consumer is built:

if (req_data.auto_offset_reset != "earliest") {
throw parse::error(
parse::error_code::invalid_param, "auto.offset must be earliest");
}
src/v/pandaproxy/rest/handlers.cc (create_consumer).

Because earliest is the only reachable policy, the value is validated and logged but never forwarded to the internal kafka::client — the client doesn't need to know it. This is what keeps recovery cheap: earliest resolves to the partition's log_start_offset, which the broker already returns in every fetch response, so the client reseeds straight from that field. No separate ListOffsets is issued to resolve the reset policy. (If the proxy ever allowed latest/timestamp, that assumption breaks and the value would have to be threaded through and resolved via ListOffsets.)

2. On offset_out_of_range we recover within the fetch, like the Confluent REST Proxy

The Confluent REST Proxy wraps the Java KafkaConsumer, whose poll() is a timer-bounded do/while: each iteration resets any out-of-range position — via a ListOffsets to the earliest offset — and refetches, returning the recovered records within the same poll if the request timeout allows, or an empty result (recovering on a later poll) if the timer expires first. consumer::fetch() mirrors that shape:

• delivers the healthy partitions immediately (their offsets advance),
• reseeds the out-of-range partitions to the broker-reported log_start_offset and strips them from the response (the pandaproxy serializer rejects any partition error),
• and, when a round only reseeded and delivered nothing, repeats the fetch round within the caller's timeout budget so the reseeded data comes back in the same fetch — exactly what the client's next poll would have done.

The observable contract matches the Confluent REST Proxy: the recovered records come back in a single fetch when the timeout allows, or empty-then-next-poll otherwise; no offset_out_of_range is ever surfaced to the client, and no already-delivered records are re-read. The one mechanical difference is offset resolution — the Java consumer issues a ListOffsets(earliest) to find the log start, we reseed straight from the log_start_offset already present in the fetch response, valid precisely because of assumption (1). An out-of-range partition sitting alongside partitions that returned records is delivered on the next poll rather than the same one: the healthy records return immediately and the reseeded partition resumes from the corrected offset, just as the Java consumer's poll() returns as soon as any partition has records.

The one case that is retried is a dispatch failure — a whole broker's response missing, where the topology may be stale. There we discard the round and throw so the retry in client::consumer_fetch() runs the update_metadata refresh a stale leader needs.

Fixes: CORE-16844

Backports Required

• none - not a bug fix
• none - this is a backport
• none - issue does not exist in previous branches
• none - papercut/not impactful enough to backport
• v26.1.x
• v25.3.x
• v25.2.x

Release Notes

Bug Fixes

• HTTP Proxy: Consumer group fetches no longer fail indefinitely with offset_out_of_range after retention moves a topic's log start offset past 0; the consumer now recovers to the earliest available offset (auto.offset.reset=earliest).

[Copilot]

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 from log_start_offset when a partition returns offset_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.

[vbotbuildovich]

Retry command for Build#86958

please wait until all jobs are finished before running the slash command

/ci-repeat 1
skip-redpanda-build
skip-units
skip-rebase
tests/rptest/tests/pandaproxy_test.py::PandaProxyConsumerGroupTest.test_consumer_group_fetch_after_prefix_trim

[vbotbuildovich]

CI test results

test results on build#86958

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(PASS)	DatalakeCustomPartitioningTest	test_many_partitions	{"catalog_type": "rest_jdbc", "cloud_storage_type": 1}	integration	https://buildkite.com/redpanda/redpanda/builds/86958#019f4c05-fcf1-4475-93fb-dfeedef9fe38	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0000, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=DatalakeCustomPartitioningTest&test_method=test_many_partitions
FAIL	PandaProxyConsumerGroupTest	test_consumer_group_fetch_after_prefix_trim	null	integration	https://buildkite.com/redpanda/redpanda/builds/86958#019f4c05-fcf2-46b3-ba2c-cee09b854379	0/11	The test was found to be new, and no failures are allowed	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=PandaProxyConsumerGroupTest&test_method=test_consumer_group_fetch_after_prefix_trim
FAIL	PandaProxyConsumerGroupTest	test_consumer_group_fetch_after_prefix_trim	null	integration	https://buildkite.com/redpanda/redpanda/builds/86958#019f4c06-23d8-4bff-b3c1-ec7712ea3e88	0/11	The test was found to be new, and no failures are allowed	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=PandaProxyConsumerGroupTest&test_method=test_consumer_group_fetch_after_prefix_trim

test results on build#87059

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(PASS)	ShadowLinkingReplicationTests	test_replication_with_failures	{"storage_mode": "tiered_v2"}	integration	https://buildkite.com/redpanda/redpanda/builds/87059#019f5fb7-443a-4315-8550-592dc1fd1220	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0000, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=ShadowLinkingReplicationTests&test_method=test_replication_with_failures

test results on build#87065

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(PASS)	ShadowLinkRoleSyncScaleTest	test_role_sync_at_scale	{"members_per_role": 1, "num_roles": 5000}	integration	https://buildkite.com/redpanda/redpanda/builds/87065#019f6081-b952-44e0-b4a1-d0a5d581fbeb	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0139, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=ShadowLinkRoleSyncScaleTest&test_method=test_role_sync_at_scale

test results on build#87391

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(INCONCLUSIVE)	NodeWiseRecoveryTest	test_node_wise_recovery	{"dead_node_count": 1}	integration	https://buildkite.com/redpanda/redpanda/builds/87391#019f8006-59e3-410a-ad3a-0cd03a5f4725	17/29	Test is INCONCLUSIVE after retries.Inconclusive result before max retries(baseline=0.0317, p0=0.0243, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.8670, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=NodeWiseRecoveryTest&test_method=test_node_wise_recovery

[bartoszpiekny-redpanda]

Force-pushed: squashed the per-partition recovery into the main fix commit so the PR shows the final design directly (no intermediate coarse-recovery version). Behaviour and net diff are unchanged from the previous push — only the commit layout.

Diff since the previous push (6d5f4d6 → a037491):
https://github.com/bartoszpiekny-redpanda/redpanda/compare/6d5f4d696d..a037491fe8

[bartoszpiekny-redpanda]

Force-push: offset_out_of_range recovery now completes within a single fetch (folded into the fix commit) + ducktape coverage for single-poll recovery. Diff vs the previous revision: https://github.com/redpanda-data/redpanda/compare/a037491fe8..f35db722ac

[pgellert]

lgtm, the main thing to consider is the timeout=0 case, the rest are minor questions

[bartoszpiekny-redpanda]

Addressed both review comments: fetch() now always runs at least one round (deadline checked after the round), so a zero or already-elapsed timeout budget still issues one non-blocking fetch instead of returning empty without trying. And reseed_out_of_range now validates log_start_offset — it dasserts the invariant (the broker always sets it alongside offset_out_of_range, see do_read_from_ntp) and, in release, warns and skips the reseed rather than seeding a negative offset and looping.

Diff of these changes: https://github.com/redpanda-data/redpanda/compare/f35db722ac..80417ceb16

[mnajda-redpanda]

lgtm!

[vbotbuildovich]

/backport v26.1.x

[vbotbuildovich]

/backport v25.3.x

[vbotbuildovich]

/backport v25.2.x