Fix consumer iteration on compacted topics

Author: dpkp

kafka/consumer/fetcher.py

@@ -438,7 +438,7 @@ def _message_generator(self):
 
                     # 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)