diff options
Diffstat (limited to 'kafka/consumer/simple.py')
| -rw-r--r-- | kafka/consumer/simple.py | 9 |
1 files changed, 8 insertions, 1 deletions
diff --git a/kafka/consumer/simple.py b/kafka/consumer/simple.py index 2c2f820..88eeada 100644 --- a/kafka/consumer/simple.py +++ b/kafka/consumer/simple.py @@ -19,7 +19,7 @@ from kafka.common import ( FetchRequest, OffsetRequest, ConsumerFetchSizeTooSmall, ConsumerNoMoreData, UnknownTopicOrPartitionError, NotLeaderForPartitionError, - OffsetOutOfRangeError, check_error + OffsetOutOfRangeError, FailedPayloadsError, check_error ) from .base import ( Consumer, @@ -355,6 +355,13 @@ class SimpleConsumer(Consumer): # Retry this partition retry_partitions[resp.partition] = partitions[resp.partition] continue + except FailedPayloadsError as e: + log.warning("Failed payloads of %s" + "Resetting partition offset...", + e.payload) + # Retry this partition + retry_partitions[e.payload.partition] = partitions[e.payload.partition] + continue partition = resp.partition buffer_size = partitions[partition] |