openstack/deb-python-kafka
Code Issues Proposed changes

Merge pull request #405 from dpkp/log_error_type

Browse Source 
Log response error types in consumer and producer logs
This commit is contained in:
Dana Powers
2015-06-10 13:19:40 -07:00
parent 4dec5d31a9 942c693c23
commit 2727a07be2
2 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
kafka/consumer/simple.py
View File

@@ -358,23 +358,26 @@ class SimpleConsumer(Consumer):
try: try:
check_error(resp) check_error(resp)
except UnknownTopicOrPartitionError: except UnknownTopicOrPartitionError:
log.error('UnknownTopicOrPartitionError for %s:%d',
resp.topic, resp.partition)
self.client.reset_topic_metadata(resp.topic) self.client.reset_topic_metadata(resp.topic)
raise raise
except NotLeaderForPartitionError: except NotLeaderForPartitionError:
log.error('NotLeaderForPartitionError for %s:%d',
resp.topic, resp.partition)
self.client.reset_topic_metadata(resp.topic) self.client.reset_topic_metadata(resp.topic)
continue continue
except OffsetOutOfRangeError: except OffsetOutOfRangeError:
log.warning("OffsetOutOfRangeError for %s - %d. " log.warning('OffsetOutOfRangeError for %s:%d. '
"Resetting partition offset...", 'Resetting partition offset...',
resp.topic, resp.partition) resp.topic, resp.partition)
self.reset_partition_offset(resp.partition) self.reset_partition_offset(resp.partition)
# Retry this partition # Retry this partition
retry_partitions[resp.partition] = partitions[resp.partition] retry_partitions[resp.partition] = partitions[resp.partition]
continue continue
except FailedPayloadsError as e: except FailedPayloadsError as e:
log.warning("Failed payloads of %s" log.warning('FailedPayloadsError for %s:%d',
"Resetting partition offset...", e.payload.topic, e.payload.partition)
e.payload)
# Retry this partition # Retry this partition
retry_partitions[e.payload.partition] = partitions[e.payload.partition] retry_partitions[e.payload.partition] = partitions[e.payload.partition]
continue continue

5
kafka/producer/base.py
View File

@@ -166,8 +166,9 @@ def _send_upstream(queue, client, codec, batch_time, batch_size,
if error_cls: if error_cls:
_handle_error(error_cls, orig_req) _handle_error(error_cls, orig_req)
log.error('Error sending ProduceRequest (#%d of %d) to %s:%d ' log.error('%s sending ProduceRequest (#%d of %d) '
'with msgs %s', i + 1, len(requests), 'to %s:%d with msgs %s',
error_cls.__name__, (i + 1), len(requests),
orig_req.topic, orig_req.partition, orig_req.topic, orig_req.partition,
orig_req.messages if log_messages_on_error orig_req.messages if log_messages_on_error
else hash(orig_req.messages)) else hash(orig_req.messages))