hubcio commented on code in PR #3525:
URL: https://github.com/apache/iggy/pull/3525#discussion_r3509371521
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -676,6 +702,13 @@ impl IggyConsumer {
sleep(retry_interval.get_duration()).await;
}
+ let effective_pid = partition_id.unwrap_or(u32::MAX);
+ let polling_strategy = if
fallback_to_last.contains_key(&effective_pid) {
+ PollingStrategy::last()
Review Comment:
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.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -530,7 +537,17 @@ impl IggyConsumer {
&self.consumer_name,
self.joined_consumer_group.clone(),
)
- .await
+ .await?;
+
+ // A brand-new consumer group starts at stored offset 0. If the topic
has
+ // had retention run, offset 0 may no longer exist. Pre-arm the
fallback
+ // so the very first poll uses PollingStrategy::last() instead of Next,
+ // avoiding a guaranteed InvalidOffset error on startup.
+ if newly_created {
Review Comment:
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 enti
rely seems right - the server already handles the fresh-group case.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -111,6 +111,12 @@ pub struct IggyConsumer {
topic_id: Arc<Identifier>,
partition_id: Option<u32>,
polling_strategy: PollingStrategy,
+ // Per-partition recovery set: when a partition's stored offset falls below
+ // its earliest available offset, its ID is inserted here so the next poll
+ // for that partition uses PollingStrategy::last(). Removed after the first
+ // successful recovery poll. Keyed by partition_id; u32::MAX is the
sentinel
+ // for consumers with no fixed partition (consumer-group auto-assign).
+ fallback_to_last: Arc<DashMap<u32, ()>>,
Review Comment:
`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.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -783,6 +817,18 @@ impl IggyConsumer {
let error = polled_messages.unwrap_err();
error!("Failed to poll messages: {error}");
+ // When the consumer group's stored offset falls below the topic's
+ // earliest available offset (e.g. after retention removes old
+ // segments), seek to the most recent message on the next poll
+ // instead of looping forever at the invalid offset.
+ if matches!(error, IggyError::InvalidOffset(_)) {
+ warn!(
+ "Consumer offset is before the earliest available message
in topic: {topic_id}, stream: {stream_id}. \
Review Comment:
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.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -783,6 +817,18 @@ impl IggyConsumer {
let error = polled_messages.unwrap_err();
error!("Failed to poll messages: {error}");
+ // When the consumer group's stored offset falls below the topic's
+ // earliest available offset (e.g. after retention removes old
+ // segments), seek to the most recent message on the next poll
+ // instead of looping forever at the invalid offset.
+ if matches!(error, IggyError::InvalidOffset(_)) {
Review Comment:
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.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -880,7 +930,10 @@ impl IggyConsumer {
return Err(error);
}
}
- }
+ true
Review Comment:
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`.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -676,6 +702,13 @@ impl IggyConsumer {
sleep(retry_interval.get_duration()).await;
}
+ let effective_pid = partition_id.unwrap_or(u32::MAX);
+ let polling_strategy = if
fallback_to_last.contains_key(&effective_pid) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
