Skip to content

Moved offset initialization to get_offsets method. #200

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

Moved offset initialization to get_offsets method. #200

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 27 additions & 13 deletions kafka/consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,20 +100,8 @@ def __init__(self, client, group, topic, partitions=None, auto_commit=True,
self.commit)
self.commit_timer.start()

def get_or_init_offset_callback(resp):
try:
kafka.common.check_error(resp)
return resp.offset
except kafka.common.UnknownTopicOrPartitionError:
return 0

if auto_commit:
for partition in partitions:
req = OffsetFetchRequest(topic, partition)
(offset,) = self.client.send_offset_fetch_request(group, [req],
callback=get_or_init_offset_callback,
fail_on_error=False)
self.offsets[partition] = offset
self.offsets = self.get_offsets(partitions)
else:
for partition in partitions:
self.offsets[partition] = 0
Expand Down Expand Up @@ -156,6 +144,31 @@ def commit(self, partitions=None):

self.count_since_commit = 0

def get_offsets(self, partitions=None):
"""
Fetch offsets for the specified partitions or all partitions. Default offset to 0 if not found.
Returns a dictionary with partition as key and the offset as value
"""
offsets = {}
if partitions is None:
partitions = self.client.topic_partitions[self.topic]

def get_or_init_offset_callback(resp):
try:
kafka.common.check_error(resp)
return resp.offset
except kafka.common.UnknownTopicOrPartitionError:
return 0
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this internal callback function should be dropped in favor of explicit error checking of the response after calling send_offset_fetch_request . If we try to get offsets for an unknown partition, shouldn't we raise an exception here? Usually that signals our metadata is messed up and a full metadata refresh is in order.


for partition in partitions:
req = OffsetFetchRequest(self.topic, partition)
(offset,) = self.client.send_offset_fetch_request(self.group, [req],
callback=get_or_init_offset_callback,
fail_on_error=False)
offsets[partition] = offset

return offsets

def _auto_commit(self):
"""
Check if we have to commit based on number of messages and commit
Expand Down Expand Up @@ -451,6 +464,7 @@ def _fetch(self):
log.debug("Done iterating over partition %s" % partition)
partitions = retry_partitions


def _mp_consume(client, group, topic, chunk, queue, start, exit, pause, size):
"""
A child process worker which consumes messages based on the
Expand Down