openstack/deb-python-kafka
Code Issues Proposed changes

Use explicit subscription state flag to handle seek() during message iteration

Browse Source
This commit is contained in:
Dana Powers
2016-07-08 15:06:37 -07:00
parent 003bb0a830
commit 7b5ade10a5
2 changed files with 17 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
kafka/consumer/fetcher.py
View File

@@ -424,6 +424,12 @@ class Fetcher(six.Iterator):
elif fetch_offset == position: elif fetch_offset == position:
log.log(0, "Returning fetched records at offset %d for assigned" log.log(0, "Returning fetched records at offset %d for assigned"
" partition %s", position, tp) " partition %s", position, tp)
# We can ignore any prior signal to drop pending message sets
# because we are starting from a fresh one where fetch_offset == position
# i.e., the user seek()'d to this position
self._subscriptions.assignment[tp].drop_pending_message_set = False
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
@@ -436,8 +442,16 @@ class Fetcher(six.Iterator):
" since it is no longer fetchable", tp) " since it is no longer fetchable", tp)
break break
# If there is a seek during message iteration,
# we should stop unpacking this message set and
# wait for a new fetch response that aligns with the
# new seek position
elif self._subscriptions.assignment[tp].drop_pending_message_set:
log.debug("Skipping remainder of message set for partition %s", tp)
self._subscriptions.assignment[tp].drop_pending_message_set = False
break
# Compressed messagesets may include earlier messages # Compressed messagesets may include earlier messages
# It is also possible that the user called seek()
elif msg.offset < self._subscriptions.assignment[tp].position: elif msg.offset < self._subscriptions.assignment[tp].position:
log.debug("Skipping message offset: %s (expecting %s)", log.debug("Skipping message offset: %s (expecting %s)",
msg.offset, msg.offset,

2
kafka/consumer/subscription_state.py
View File

@@ -350,6 +350,7 @@ class TopicPartitionState(object):
self.reset_strategy = None # the reset strategy if awaitingReset is set self.reset_strategy = None # the reset strategy if awaitingReset is set
self._position = None # offset exposed to the user self._position = None # offset exposed to the user
self.highwater = None self.highwater = None
self.drop_pending_message_set = False
def _set_position(self, offset): def _set_position(self, offset):
assert self.has_valid_position, 'Valid position required' assert self.has_valid_position, 'Valid position required'
@@ -371,6 +372,7 @@ class TopicPartitionState(object):
self.awaiting_reset = False self.awaiting_reset = False
self.reset_strategy = None self.reset_strategy = None
self.has_valid_position = True self.has_valid_position = True
self.drop_pending_message_set = True
def pause(self): def pause(self):
self.paused = True self.paused = True