try to fix uncaught FailedPayloadsError

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]