diff options
| author | Dana Powers <dana.powers@gmail.com> | 2015-03-02 17:36:48 -0800 |
|---|---|---|
| committer | Dana Powers <dana.powers@gmail.com> | 2015-03-02 17:36:48 -0800 |
| commit | 7dea5599b64e6e59718b35eb2ce423c12b283799 (patch) | |
| tree | 69f37452bf99144b0827e64dfd76d9ee56e2ace5 /kafka/consumer/kafka.py | |
| parent | 2ca10e26b6fe00943db6984d30d5d144da76629e (diff) | |
| parent | 6de9444b960b931d9def09e6d95ac6866bb34466 (diff) | |
| download | kafka-python-7dea5599b64e6e59718b35eb2ce423c12b283799.tar.gz | |
Merge pull request #328 from wkiser/validate_offset
Fixes consumer/kafka and consumer/simple to only yield messages if the m...
Diffstat (limited to 'kafka/consumer/kafka.py')
| -rw-r--r-- | kafka/consumer/kafka.py | 4 |
1 files changed, 4 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 |