diff options
| author | wkiser <wkiser@gmail.com> | 2015-02-23 12:58:20 -0500 |
|---|---|---|
| committer | wkiser <wkiser@gmail.com> | 2015-03-02 13:43:53 -0500 |
| commit | 6de9444b960b931d9def09e6d95ac6866bb34466 (patch) | |
| tree | 446691376535a3f33463284d1561d6c5d7423fe8 /kafka/consumer/simple.py | |
| parent | 9ad0be662d388b47aadf04d712f5744add6456e3 (diff) | |
| download | kafka-python-6de9444b960b931d9def09e6d95ac6866bb34466.tar.gz | |
Fixes consumer/kafka and consumer/simple to only yield messages if the message's offset is greater than or equal to the consumer offset.
Diffstat (limited to 'kafka/consumer/simple.py')
| -rw-r--r-- | kafka/consumer/simple.py | 4 |
1 files changed, 4 insertions, 0 deletions
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 |