openstack/deb-python-kafka
Code Issues Proposed changes

Fix offset increments:

Browse Source 
* Increment the offset before returning a message rather than when 
  putting it in the internal queue. This prevents committing the wrong offsets.
* In MultiProcessConsumer, store the offset of the next message
This commit is contained in:
Omar Ghishan
2014-01-15 11:13:59 -08:00
parent 9644166048
commit c36cb618d2
1 changed files with 17 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
kafka/consumer.py
View File

@@ -338,7 +338,19 @@ class SimpleConsumer(Consumer):
with FetchContext(self, block, timeout):
self._fetch()
try:
return self.queue.get_nowait()
partition, message = self.queue.get_nowait()
# Update partition offset
self.offsets[partition] = message.offset + 1
# Count, check and commit messages if necessary
self.count_since_commit += 1
self._auto_commit()
if self.partition_info:
return partition, message
else:
return message
except Empty:
return None
@@ -380,18 +392,8 @@ class SimpleConsumer(Consumer):
partition = resp.partition
try:
for message in resp.messages:
# Update partition offset
self.offsets[partition] = message.offset + 1
# Count, check and commit messages if necessary
self.count_since_commit += 1
self._auto_commit()
# Put the message in our queue
if self.partition_info:
self.queue.put((partition, message))
else:
self.queue.put(message)
except ConsumerFetchSizeTooSmall, e:
if (self.max_buffer_size is not None and
self.buffer_size == self.max_buffer_size):
@@ -577,12 +579,11 @@ class MultiProcessConsumer(Consumer):
break
# Count, check and commit messages if necessary
self.offsets[partition] = message.offset
self.offsets[partition] = message.offset + 1
self.start.clear()
yield message
self.count_since_commit += 1
self._auto_commit()
yield message
self.start.clear()
@@ -624,7 +625,7 @@ class MultiProcessConsumer(Consumer):
messages.append(message)
# Count, check and commit messages if necessary
self.offsets[partition] = message.offset
self.offsets[partition] = message.offset + 1
self.count_since_commit += 1
self._auto_commit()
count -= 1