From 945fc048a8cc61e1a9390bd7a7fed371d2e23277 Mon Sep 17 00:00:00 2001
From: reAsOn2010 <the.reason.sake@gmail.com>
Date: Tue, 9 Jun 2015 04:20:16 +0800
Subject: [PATCH] try to fix uncaught FailedPayloadsError

---
 kafka/common.py                   | 2 ++
 kafka/consumer/simple.py          | 9 ++++++++-
 test/test_failover_integration.py | 8 ++++++++
 3 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/kafka/common.py b/kafka/common.py
index 2fdf7d2ce..66987ffe2 100644
--- a/kafka/common.py
+++ b/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)
diff --git a/kafka/consumer/simple.py b/kafka/consumer/simple.py
index 2c2f820ec..88eeada25 100644
--- a/kafka/consumer/simple.py
+++ b/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]
diff --git a/test/test_failover_integration.py b/test/test_failover_integration.py
index f26009373..1d835e25b 100644
--- a/test/test_failover_integration.py
+++ b/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):