Repository navigation
Release fetch_context when the fetch raises - #1190
Open
kratos0718 wants to merge 3 commits into
Open
kratos0718 wants to merge 3 commits into
kratos0718 wants to merge 3 commits into
Conversation
getone() and getmany() await inside fetch_context(). If that await raises, most commonly CancelledError from asyncio.wait_for() timing out, the fetch counter was never decremented, so fetcher_idle_time reported 0 from then on and the max_poll_interval check never made the consumer leave the group.
 |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
This comment has been minimized.
This comment has been minimized.
This branch has not been deployed
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
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.
getone()andgetmany()await insideSubscriptionState.fetch_context(), but the context manager decremented_fetch_countafter a bareyield. If the await raises, most commonlyCancelledErrorwhen the call is wrapped inasyncio.wait_for()and times out, the counter stays incremented. From then onfetcher_idle_timereturns 0, so the heartbeat loop'smax_poll_intervalcheck never makes the consumer leave the group, and the rejoin path treats it as active.I reproduced this against a local Kafka 3.8 broker. The consumer used
max_poll_interval_ms=3000, made oneasyncio.wait_for(consumer.getone(), timeout=1)call that timed out, then stopped polling for 8 seconds. On master it never sent LeaveGroup. With this change it leaves as expected.The fix wraps the
yieldintry/finally. I added a unit test intests/test_subscription_state.pythat fails on master (idle time stays 0) and passes with the change.make lintis clean. I also added a CHANGES.rst entry.