Merge pull request #392 from reAsOn2010/fix-uncaught-FailedPayloadsError

try to fix uncaught FailedPayloadsError in consumer

Author: dpkp

kafka/common.py

@@ -226,6 +226,8 @@ def _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)

kafka/consumer/simple.py

@@ -19,7 +19,7 @@
     FetchRequest, OffsetRequest,
     ConsumerFetchSizeTooSmall, ConsumerNoMoreData,
     UnknownTopicOrPartitionError, NotLeaderForPartitionError,
-    OffsetOutOfRangeError, check_error
+    OffsetOutOfRangeError, FailedPayloadsError, check_error
 )
 from .base import (
     Consumer,
@@ -355,6 +355,13 @@ def _fetch(self):
                     # 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]

test/test_failover_integration.py

@@ -169,6 +169,14 @@ def test_switch_leader_keyed_producer(self):
             msg = random_string(10).encode('utf-8')
             producer.send_messages(topic, key, msg)
 
+    @kafka_versions("all")
+    def test_switch_leader_simple_consumer(self):
+        producer = Producer(self.client, async=False)
+        consumer = SimpleConsumer(self.client, None, self.topic, partitions=None, auto_commit=False, iter_timeout=10)
+        self._send_random_messages(producer, self.topic, 0, 2)
+        consumer.get_messages()
+        self._kill_leader(self.topic, 0)
+        consumer.get_messages()
 
     def _send_random_messages(self, producer, topic, partition, n):
         for j in range(n):