openstack/deb-python-kafka
Code Issues Proposed changes

Fetcher iterator should check for pause and seek resets

Browse Source
This commit is contained in:
Dana Powers
2016-01-31 23:42:04 -08:00
parent 130155b874
commit 33b9ff2e33
1 changed files with 5 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
kafka/consumer/fetcher.py
View File

@@ -400,11 +400,13 @@ class Fetcher(six.Iterator):
for msg in self._unpack_message_set(tp, messages): for msg in self._unpack_message_set(tp, messages):
# Because we are in a generator, it is possible for # 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 # 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" log.debug("Not returning fetched records for partition %s"
" since it is no longer assigned", tp) " since it is no longer fetchable", tp)
break break
# Compressed messagesets may include earlier messages # Compressed messagesets may include earlier messages