diff options
Diffstat (limited to 'kafka/consumer.py')
| -rw-r--r-- | kafka/consumer.py | 5 |
1 files changed, 4 insertions, 1 deletions
diff --git a/kafka/consumer.py b/kafka/consumer.py index 1404097..8ac28da 100644 --- a/kafka/consumer.py +++ b/kafka/consumer.py @@ -275,7 +275,6 @@ class SimpleConsumer(Consumer): 2 is relative to the latest known offset (tail) """ - self.count_since_commit += 1 if whence == 1: # relative to current position for partition, _offset in self.offsets.items(): self.offsets[partition] = _offset + offset @@ -306,6 +305,10 @@ class SimpleConsumer(Consumer): # Reset queue and fetch offsets since they are invalid self.fetch_offsets = self.offsets.copy() + if self.auto_commit: + self.count_since_commit += 1 + self.commit() + self.queue = Queue() def get_messages(self, count=1, block=True, timeout=0.1): |