fix(sdk): stream should terminate after consumer shutdown - #3964
Merged
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3964 +/- ##
============================================
- Coverage 84.97% 84.94% -0.04%
Complexity 1402 1402
============================================
Files 1225 1225
Lines 180182 180224 +42
Branches 146486 146528 +42
============================================
- Hits 153107 153088 -19
- Misses 23043 23061 +18
- Partials 4032 4075 +43
🚀 New features to boost your workflow:
|
spetz
approved these changes
Sep 1, 2026
hubcio
approved these changes
Sep 1, 2026
hubcio
added a commit
to haubur/iggy
that referenced
this pull request
Sep 1, 2026
The IggyConsumer rustdoc added on this branch promises behaviour the SDK does not have, and three of those promises lose data when followed. Any polling strategy but next() on a group member: the consumer keeps one continuation cursor for all partitions, applies it to whichever partition the next poll hits, and the default auto-commit commits the skipped range. AutoCommit::Disabled with shutdown(): the final flush commits the reading position under every setting, so the message whose handler failed goes with it. store_offset(0) as a rewind: next() resumes at stored + 1 and the interval task re-commits the old position within a second. Rewrite the affected sections against the code at this revision. The stream yields None after shutdown() since apache#3964, AutoCommitAfter fires at the partition head seen by the poll and also after Err, a failed decrypt is a permanent skip under the default and a retry loop otherwise, partition_id() reflects the last poll response including empty ones, the offset getters seed Some(0), surplus group members keep syncing instead of idling, and do_not_auto_join_consumer_group() leaves a member unable to poll. The examples match on the stream item instead of bailing out with ?, and the manual-commit example drops the inert allow_replay(). Restore the apache#3668 comments that explain why the commit tasks drain before the group leave, cut comments that restate the next line, and keep the auto-commit mechanics in one place, the type-level doc, with init(), shutdown() and the Stream impl linking to it. Doc and comment lines only. The code defects behind these sentences are left for separate fixes.
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
4 participants
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.
Which issue does this PR address?
Closes #3910
Rationale
After
consumer.shutdown()calling consumer.next() was still possible. CausingWhat changed?
When writing the issue I said that the preferred way would be to return an IggyError to allow user handling.
However, calling next() on a shut down consumer now returns:
Poll::Ready(None)indicating a terminated stream as intended by the futures crate: https://docs.rs/futures/latest/futures/prelude/trait.Stream.html#:~:text=Poll%3A%3AReady%28None%29%20means%20that%20the%20stream%20has%20terminated%2C%20and%20poll%5Fnext%20should%20not%20be%20invoked%20againLocal Execution