From 6de9444b960b931d9def09e6d95ac6866bb34466 Mon Sep 17 00:00:00 2001 From: wkiser Date: Mon, 23 Feb 2015 12:58:20 -0500 Subject: [PATCH] Fixes consumer/kafka and consumer/simple to only yield messages if the message's offset is greater than or equal to the consumer offset. --- kafka/consumer/kafka.py | 4 ++++ kafka/consumer/simple.py | 4 ++++ 2 files changed, 8 insertions(+) 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 000fcd9..4f76bd6 100644 --- a/kafka/consumer/simple.py +++ b/kafka/consumer/simple.py @@ -305,6 +305,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