Merge pull request #328 from wkiser/validate_offset

Fixes consumer/kafka and consumer/simple to only yield messages if the m...

1 files changed, 4 insertions, 0 deletions

diff --git a/kafka/consumer/simple.py b/kafka/consumer/simple.py
index 4c835fe..3d250ea 100644
--- a/kafka/consumer/simple.py
+++ b/kafka/consumer/simple.py
@@ -356,6 +356,10 @@ class SimpleConsumer(Consumer):
                 buffer_size = partitions[partition]
                 try:
                     for message in resp.messages:
+                        if message.offset < self.fetch_offsets[partition]:
+                            log.debug('Skipping message %s because its offset is less than the consumer offset',
+                                      message)
+                            continue
                         # Put the message in our queue
                         self.queue.put((partition, message))
                         self.fetch_offsets[partition] = message.offset + 1