Skip to content
This repository was archived by the owner on Mar 24, 2021. It is now read-only.

leaderless topics #230

Merged
merged 3 commits into from
Aug 21, 2015
Merged

leaderless topics #230

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
53c172d
don't return None when topic returns a Leader error - they will be ca…
emmettbutler Aug 14, 2015
50820dd
Merge branch 'master' into bugfix/leaderless-topics
emmettbutler Aug 21, 2015
14efb7f
retry in response to LeaderNotAvailable
emmettbutler Aug 21, 2015
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
12 changes: 4 additions & 8 deletions pykafka/broker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -282,18 +282,14 @@ def request_metadata(self, topics=None):
future = self._req_handler.request(MetadataRequest(topics=topics))
response = future.get(MetadataResponse)

errored = False
for name, topic_metadata in response.topics.iteritems():
if topic_metadata.err == LeaderNotAvailable.ERROR_CODE:
log.warning("Leader not available.")
errored = True
log.warning("Leader not available for topic '%s'.", name)
for pid, partition_metadata in topic_metadata.partitions.iteritems():
if partition_metadata.err == LeaderNotAvailable.ERROR_CODE:
log.warning("Leader not available.")
errored = True

if not errored:
return response
log.warning("Leader not available for topic '%s' partition %d.",
name, pid)
return response

######################
# Commit/Fetch API #
Expand Down
4 changes: 4 additions & 0 deletions pykafka/producer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from .exceptions import (
InvalidMessageError,
InvalidMessageSize,
LeaderNotAvailable,
MessageSizeTooLarge,
NotLeaderForPartition,
ProduceFailureError,
Expand Down Expand Up @@ -284,6 +285,9 @@ def _get_partition_msgs(partition_id, req):
log.warning('Produce request to %s:%s timed out. '
'Retrying.', owned_broker.broker.host,
owned_broker.broker.port)
elif presponse.err == LeaderNotAvailable.ERROR_CODE:
log.warning('Leader not available for partition %s.'
'Retrying.', partition)
elif presponse.err == InvalidMessageError.ERROR_CODE:
log.warning('Encountered InvalidMessageError')
elif presponse.err == InvalidMessageSize.ERROR_CODE:
Expand Down