From 33b9ff2e339db34684ba0189e1f4e5865bc4ed9f Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 31 Jan 2016 23:42:04 -0800 Subject: [PATCH] Fetcher iterator should check for pause and seek resets --- kafka/consumer/fetcher.py | 8 +++++--- 1 file 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