diff --git a/kafka/consumer/kafka.py b/kafka/consumer/kafka.py
index cd3cc4a2b..53ba0a722 100644
--- a/kafka/consumer/kafka.py
+++ b/kafka/consumer/kafka.py
@@ -430,6 +430,10 @@ def fetch_messages(self):
                                    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 000fcd97b..4f76bd6d5 100644
--- a/kafka/consumer/simple.py
+++ b/kafka/consumer/simple.py
@@ -305,6 +305,10 @@ def _fetch(self):
                 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