Merge pull request #328 from wkiser/validate_offset

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

2 files changed, 8 insertions, 0 deletions

diff --git a/kafka/consumer/kafka.py b/kafka/consumer/kafka.py
index cd3cc4a..53ba0a7 100644
--- a/kafka/consumer/kafka.py
+++ b/kafka/consumer/kafka.py
@@ -430,6 +430,10 @@ class KafkaConsumer(object):
                                    offset, message.key,
                                    self._config['deserializer_class'](message.value))
 
+                if offset < self._offsets.fetch[topic_partition]:
+                    logger.debug('Skipping message %s because its offset is less than the consumer offset',
+                                 msg)
+                    continue
                 # Only increment fetch offset if we safely got the message and deserialized
                 self._offsets.fetch[topic_partition] = offset + 1
 
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