try to fix uncaught FailedPayloadsError

2 files changed, 10 insertions, 1 deletions

diff --git a/kafka/common.py b/kafka/common.py
index 2fdf7d2..66987ff 100644
--- a/kafka/common.py
+++ b/kafka/common.py
@@ -226,6 +226,8 @@ kafka_errors = dict([(x.errno, x) for x in _iter_broker_errors()])
 
 
 def check_error(response):
+    if isinstance(response, Exception):
+        raise response
     if response.error:
         error_class = kafka_errors.get(response.error, UnknownError)
         raise error_class(response)
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]