From e5af3170f3f7fb095526269ff20e897ab08c0015 Mon Sep 17 00:00:00 2001 From: Mark Roberts Date: Tue, 25 Mar 2014 11:07:31 -0700 Subject: [PATCH 1/2] Make seek(); commit(); work without commit discarding the seek change --- kafka/consumer.py | 1 + 1 file changed, 1 insertion(+) diff --git a/kafka/consumer.py b/kafka/consumer.py index 28b53ec92..140409721 100644 --- a/kafka/consumer.py +++ b/kafka/consumer.py @@ -275,6 +275,7 @@ def seek(self, offset, whence): 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 From 7d425d07c26b7900aae741a7b89c7001907b73a9 Mon Sep 17 00:00:00 2001 From: Mark Roberts Date: Thu, 27 Mar 2014 10:26:52 -0700 Subject: [PATCH 2/2] Commit in seek if autocommit --- kafka/consumer.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/kafka/consumer.py b/kafka/consumer.py index 140409721..8ac28daf4 100644 --- a/kafka/consumer.py +++ b/kafka/consumer.py @@ -275,7 +275,6 @@ def seek(self, offset, whence): 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 @@ def seek(self, offset, whence): # 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):