Merge pull request #752 from dpkp/compacted_offsets

Fix consumer iteration on compacted topics

Author: dpkp

kafka/consumer/fetcher.py

@@ -424,6 +424,12 @@ def _message_generator(self):
             elif fetch_offset == position:
                 log.log(0, "Returning fetched records at offset %d for assigned"
                            " 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):
 
                     # Because we are in a generator, it is possible for
@@ -436,9 +442,17 @@ def _message_generator(self):
                                   " since it is no longer fetchable", tp)
                         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
-                    # 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)",
                                   msg.offset,
                                   self._subscriptions.assignment[tp].position)

kafka/consumer/subscription_state.py

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