diff options
| -rw-r--r-- | kafka/consumer/fetcher.py | 8 |
1 files changed, 5 insertions, 3 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index e2bc892..41f53aa 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -400,11 +400,13 @@ class Fetcher(six.Iterator): for msg in self._unpack_message_set(tp, messages): # Because we are in a generator, it is possible for - # assignment to change between yield calls + # subscription state to change between yield calls # so we need to re-check on each loop - if not self._subscriptions.is_assigned(tp): + # this should catch assignment changes, pauses + # and resets via seek_to_beginning / seek_to_end + if not self._subscriptions.is_fetchable(tp): log.debug("Not returning fetched records for partition %s" - " since it is no longer assigned", tp) + " since it is no longer fetchable", tp) break # Compressed messagesets may include earlier messages |