[CORE-14223] kafka/client: full fetch when broker withholds an unknown partition

[mnajda-redpanda]

A replicator restart (leadership change or partition move) recreates the
direct_consumer subscription with default-initialized source offsets. The partition
goes into one fetch request, and if the source partition is already drained the
broker's incremental fetch filter omits it - so update_and_filter_offsets() never
runs and the source offsets stay at -1 indefinitely. Per KIP-227 the broker omits
unchanged partitions from incremental fetch responses; a full fetch is returned
unfiltered.

Replication itself is not affected - a partition omitted under KIP-227 is still
subscribed. What breaks is reporting: GetShadowTopic returns source_high_watermark = -1, redpanda_shadow_link_shadow_lag is wrong, and rpk shadow status prints a
negative lag. This is also the cause of the ShadowLinkingReplicationTests flake,
which compares source_high_watermark against the source topic's high watermark.

The fix tracks whether the broker has ever reported each partition, and resets the
fetch session to force a full fetch when it has not. Full fetches are returned
unpruned, so this converges in one extra fetch cycle. The reset is fetcher-wide, so
one unknown partition re-sends every partition on that broker - bounded to a single
cycle, and only on subscription recreation.

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.2.x
• v26.1.x
• v25.3.x

Release Notes

• none

[Copilot]

Pull request overview

Fixes a direct-consumer fetch-session edge case where a recreated subscription can end up with “unknown” source offsets (e.g., source_high_watermark = -1) indefinitely when the broker withholds an unchanged partition in incremental fetch responses (KIP-227). The change ensures the client forces a one-time full fetch when it detects a partition has never been reported by the broker.

Changes:

• Track whether each partition has ever been reported by the broker (source_reported) and request a fetch-session reset to force a full fetch when needed.
• Plumb a needs_full_fetch signal through fetch processing and reset the session after processing the triggering response.
• Add a regression test covering subscription teardown/recreate while the source partition is idle.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
src/v/kafka/client/direct_consumer/tests/direct_consumer_fixture_test.cc	Adds a regression test for learning source offsets after an idle reassign/restart scenario.
src/v/kafka/client/direct_consumer/fetcher.h	Extends per-partition fetch state with a source_reported flag and adds needs_full_fetch to fetch results.
src/v/kafka/client/direct_consumer/fetcher.cc	Implements detection of withheld “never reported” partitions and forces a full fetch by resetting the fetch session.

[vbotbuildovich]

CI test results

test results on build#87775

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(PASS)	NodeWiseRecoveryTest	test_recovery_local_data_missing	{"wait_for_final_manifest_uploads": false}	integration	https://buildkite.com/redpanda/redpanda/builds/87775#019fa457-4938-4391-adef-27e6779c1c4a	19/21	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0172, p0=0.2934, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3917, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=NodeWiseRecoveryTest&test_method=test_recovery_local_data_missing