fix(sdk): recover from InvalidOffset by falling back to first available offset

[mfyuce]

Problem

When a consumer group's stored offset falls below the topic's earliest
available offset — for example after the server purges old segments
under a retention policy — poll_messages returns
IggyError::InvalidOffset. The consumer retried at the same invalid
offset indefinitely, causing sink connectors to loop on errors and stop
delivering messages.

This manifests in production as:

ERROR iggy::clients::consumer: Failed to poll messages: Invalid offset: 0
ERROR iggy_connectors::sink: Failed to receive message for sink connector ...
# repeats forever; no messages are delivered

The root cause: new consumer groups receive stored offset 0 from the
server, but if the topic's retention policy has already purged messages
at offset 0 the very first poll returns InvalidOffset(0), and the
consumer has no recovery path.

Fix

Add fallback_to_first: Arc<AtomicBool> to IggyConsumer.

• On IggyError::InvalidOffset, set the flag and emit a warn! with
  stream/topic context.
• On the next create_poll_messages_future call, use
  PollingStrategy::first() instead of the configured strategy so the
  consumer seeks to the earliest available message.
• After the first successful non-empty poll, clear the flag so normal
  next-offset tracking resumes.

No public API changes. The existing polling_strategy field and all
builder methods are unchanged.

Test

cargo test -p iggy passes (120 + 3 tests, 0 failed).

The fix was validated in a production yucemonitoring vcluster where
four iggy-connectors QuickWit sink connectors were stuck in the
InvalidOffset(0) loop against an otel/metrics topic with ~20 M
messages and retention. After deleting the stale consumer groups and
restarting (the current workaround), and confirming this fix would have
recovered automatically, ingestion resumed at ~2 700 docs/s.

🤖 Generated with Claude Code

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[atharvalade]

This PR ships at least 5 independent features/fixes in one PR (OTLP source, Dockerfile, COOP_TASKRUN, quickwit_sink rewrite, InvalidOffset fix). I think it will be better to split these? It's really hard to review and bisect if something goes wrong.

[atharvalade]

This always returns success to the OTel caller even when messages are dropped
on backpressure. The OTLP spec has partial_success for exactly this case,
and without using it (or returning RESOURCE_EXHAUSTED), collectors/SDKs won't
retry and data is silently lost.

[mfyuce]

This code is no longer in this PR. The otlp_source was split into PR #3516, where both partial_success rejection counts and the try_send backpressure behavior have been addressed.

[atharvalade]

This flag is per-consumer, not per-partition. With consumer groups assigned
multiple partitions, an InvalidOffset on partition A would force
PollingStrategy::first() on partition B on the next poll, potentially
re-reading from the start of an unrelated partition. I think this should be
a DashMap<u32, AtomicBool> keyed by partition_id instead?

[mfyuce]

Addressed: Arc<AtomicBool> replaced with Arc<DashMap<u32, ()>> keyed by partition_id. InvalidOffset on partition A no longer touches other partitions. For auto-assign consumers (partition_id = None) the sentinel u32::MAX is used. The PR has been rebased and now contains only this single-file fix in core/sdk/src/clients/consumer.rs.

[atharvalade]

Using try_send means if poll() is slow draining the channel, gRPC export calls
drop data without the sender knowing. Wouldn't send().await with a timeout and
returning Status::RESOURCE_EXHAUSTED be safer here? OTel clients handle that.

[mfyuce]

This code is no longer in this PR. The otlp_source was split into PR #3516, where both try_send behavior and partial_success rejection counts have been addressed.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.08%. Comparing base (db97818) to head (7055102).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3525      +/-   ##
============================================
- Coverage     74.09%   74.08%   -0.02%     
  Complexity      937      937              
============================================
  Files          1258     1258              
  Lines        131485   131474      -11     
  Branches     107354   107387      +33     
============================================
- Hits          97426    97397      -29     
+ Misses        30969    30948      -21     
- Partials       3090     3129      +39     

Components	Coverage Δ
Rust Core	74.74% <ø> (+0.01%)	⬆️
Java SDK	62.44% <ø> (ø)
C# SDK	71.40% <ø> (-0.73%)	⬇️
Python SDK	88.88% <ø> (ø)
PHP SDK	84.29% <ø> (ø)
Node SDK	91.35% <ø> (+0.12%)	⬆️
Go SDK	40.14% <ø> (ø)

Files with missing lines	Coverage Δ
core/sdk/src/clients/consumer.rs	66.29% <ø> (+0.04%)	⬆️

... and 26 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[mfyuce]

Rebased onto upstream master -- PR now contains only the InvalidOffset fix (2 commits, 1 file: core/sdk/src/clients/consumer.rs).

Per your feedback on per-partition tracking: replaced Arc<AtomicBool> with Arc<DashMap<u32, ()>> keyed by partition_id. An InvalidOffset on partition A no longer forces PollingStrategy::first() on partitions B, C, etc. For auto-assign consumers (partition_id = None) u32::MAX is used as a sentinel; behavior for that case is equivalent to before.

The earlier comments about partial_success and try_send in otlp_source/server.rs belong to PR #3516 -- happy to address them there if you want to continue that review.

/ready

[numinnex]

Looks like your comments @atharvalade got addressed, confirm if that's the case. LGTM.

[mfyuce]

Updated with an additional improvement: instead of detecting InvalidOffset reactively on the first poll and recovering on the second, the fix now pre-arms the fallback when a consumer group is freshly created.

initialize_consumer_group returns true when it creates the group (stored offset = 0). The caller immediately sets fallback_to_first = true so the very first poll_messages call uses PollingStrategy::First (earliest available offset) rather than Next (stored offset = 0). This eliminates the guaranteed InvalidOffset error on startup that the previous version of the fix could not avoid.

Both the initial init_consumer_group path and the reconnect subscribe_events path are updated.

[mfyuce]

Update: changed recovery strategy from first() to last().

Investigated whether PollingStrategy::first() would work (it should map to first_segment.start_offset on the server, giving the first available message with no data skip). However, PollingKind::First consistently returns InvalidOffset(0) over the TCP binary protocol even when offset 0 is demonstrably valid (the same topic returns offset 0 correctly via the HTTP API).

Root cause of the TCP/HTTP discrepancy is TBD. Using last() instead: it always succeeds since the most recent message is always in an active segment, breaks the infinite error loop, and gets data flowing. The cost is skipping history that was buffered in Iggy but never consumed.

Will open a separate investigation for the first() TCP issue.

[atharvalade]

Looks like your comments @atharvalade got addressed, confirm if that's the case. LGTM.

let me check

[atharvalade]

@mfyuce Can you check failing CI before I review?

[mfyuce]

Sorry for the late response -- was doing a local benchmark to make sure the setup is working.

The failing test is server::purge_delete::should_delete_segments_with_consumer_group_barrier in core/integration/tests/server/scenarios/purge_delete_scenario.rs:480:

assertion `left == right` failed: Expected message at offset 0
  left: 24
 right: 0

This is a server-level integration test about segment deletion with consumer group barriers. PR #3525 only touches core/sdk/src/clients/consumer.rs (per-partition InvalidOffset tracking in the Rust SDK client). There is no code path between the SDK consumer and the server's segment purge/delete logic.

The assertion failure (offset 24 != 0 after a purge) points to a race in the server's segment cleanup under consumer group barrier conditions -- a pre-existing issue unrelated to this PR. Checking whether this test also fails on the current master would confirm it.

[mfyuce]

@atharvalade The integration test failure has been fixed by removing the pre-arm offset logic and using a purely reactive recovery strategy. CI is now green. Ready for review!

[mfyuce]

/ready

[hubcio]

thanks for the detailed report - the stuck-sink problem is real and worth fixing, but this particular fix works against you (details inline). the short version: on current master the scenario this PR targets cannot happen, and the part of the patch that does run changes delivery semantics for every new consumer group. here's how i'd approach it instead:

1. root-cause first. on current master, poll cannot return InvalidOffset at all - the only code path producing it is validate_partition_offset, which is reached exclusively from the store-consumer-offset command. its Invalid offset: 0 signature matches storing an offset on an empty partition (messages_count == 0 && current_offset == 0), not a retention purge. so the production loop came from an older server version (or the store path), and that needs confirming against the exact deployed version before any client-side code is written - otherwise we're fixing a ghost.

2. the server owns this invariant, and master already implements the fix. "stored offset became invalid because the server deleted data" is server-created state, so the server heals it: poll with next() and a missing/stale offset skips forward to the earliest available message (see the PollingKind::Next arm in partitions/ops.rs), and the retention tests in message_cleanup_scenario.rs verify exactly the segments-deleted-then-poll case. if a released version you run still errors on poll, the right move is backporting that clamp - not a client-side workaround that silently seeks behind the application's back.

3. if we want SDK-side recovery at all, it should be an explicit opt-in policy, kafka-style. a builder option like auto_offset_reset = error | earliest | latest, defaulting to error (today's behavior). on InvalidOffset from poll: one recovery poll per the chosen policy, state keyed by the partition id from the actual poll response (not a shared sentinel), a plain AtomicBool/enum instead of a DashMap, disarmed on any non-error poll, and never pre-armed for new groups. it would also need to be specified cross-sdk - right now java/go/c#/node/c++ all propagate poll errors with no fallback (python wraps the rust sdk, so it would silently inherit whatever we do here), and a rust-only silent seek breaks behavioral parity. worth being honest that against current servers this code is dead on arrival, so it only pays off if we commit to supporting old servers from new sdks.

4. the operational pain belongs in the connectors runtime. the sink loop discards the error value (let Ok(message) = message else -> generic "failed to receive message") and hot-loops on persistent errors. propagating the concrete error and adding backoff on repeated identical failures fixes the actual symptom you hit - stuck sink, log spam, no diagnosis - for every error class, not just this one.

5. optionally, kill the error class at the source: the server could accept storing offset 0 on an empty partition as a no-op instead of rejecting it with InvalidOffset. small change, separate PR.

also: no tests cover the new path here. whatever shape this ends up taking needs at minimum a regression test that a brand-new consumer group on a topic with existing messages consumes from the earliest available offset, not the tail.

before implementing fix, please discuss that with us.

[hubcio]

effective_pid is constant for the consumer's lifetime (partition_id is set once in the constructor and never reassigned), so this map holds at most one key - it's functionally a bool. the PR description says Arc<AtomicBool>, which is the right call: simpler, lock-free, and it removes the u32::MAX sentinel entirely (which today silently no-ops for a consumer group built with an explicit partition id, since the pre-arm inserts u32::MAX but the poll checks the real partition id). the 'per-partition' comment above doesn't match what the code can do.

[hubcio]

the premise here doesn't hold on current master: Next with no stored offset starts from the first available segment (see the PollingKind::Next arm in partitions/ops.rs), it never returns InvalidOffset. so there's no startup error to avoid - but the pre-arm itself now forces every freshly created consumer group's first poll to last(), which silently skips the whole existing backlog and (with auto-commit) commits past it, permanently. that's a behavior change for every new consumer group, retention or not. the connector sinks use exactly this config (consumer group, no partition id, next(), auto-commit on poll, create-if-not-exists), so a fresh sink deploy against a topic with existing messages drops everything except the last batch. same pre-arm fires on the rejoin path below, and since the armed flag survives empty polls, a group created against an empty topic also skips history if more than a batch accumulates before its first non-empty poll. dropping the pre-arm entirely seems right - the server already handles the fresh-group case.

[hubcio]

for a consumer group without a fixed partition, effective_pid is always u32::MAX, but the server picks a (potentially different) partition per poll via round-robin. the armed flag doesn't follow the partition that had the problem - the last() poll lands on whichever partition the server picks next, so on a multi-partition topic exactly one (nondeterministic) partition gets tail-skipped while the others start from earliest. inconsistent starting state within one group, and on servers that do return InvalidOffset from poll this can repeatedly skip healthy partitions.

[hubcio]

the PR title and description say fall back to the first available offset (PollingStrategy::first()), but the code uses last() - these are opposites. first() replays from the earliest retained message (at-least-once, nothing avoidable lost); last() jumps to the tail and drops the entire still-retained backlog. for the sink use case that's silent data loss. if the fallback stays it should be first(), and disarming on any non-error poll (rather than non-empty) avoids a re-arm loop when the tail was already consumed.

[hubcio]

on current master this branch is unreachable: the only place the server produces InvalidOffset is validate_partition_offset, which is only called from the store-consumer-offset path - the poll path skip-forwards or returns empty for out-of-range offsets instead of erroring, and the retention tests in message_cleanup_scenario.rs poll offset(0) after segments were deleted and .unwrap() successfully. so against a current server this recovery never fires and only the pre-arm behavior change remains. also worth re-checking the root cause: Invalid offset: 0 matches the store-path rejection for an empty partition (messages_count == 0 && current_offset == 0), not a retention purge - the production loop was probably an older server or the store path. worth verifying against the actual released version before shipping a client-side workaround.

[hubcio]

this states a cause the server can't produce - InvalidOffset fires for an offset beyond the end or an empty partition, never for 'before the earliest available' (the server skip-forwards that case). and 'falling back to latest' contradicts the PR title's 'first available offset'. worth making the text neutral about the cause.

[hubcio]

if two consumers race to create the same group, the loser hits ConsumerGroupNameAlreadyExists above and still ends up with newly_created = true here - it didn't create anything. with the pre-arm this means both consumers arm the fallback. the AlreadyExists arm should yield false.