fix exception when messageset is too large for decode buffer

new exception does not inherit from ProtocolClientError, removing ambiguity about whether this error originated at the broker or the client

fixes #697

pykafka/exceptions.py

@@ -51,6 +51,11 @@ class NoMessagesConsumedError(KafkaException):
     pass
 
 
+class MessageSetDecodeFailure(KafkaException):
+    """Indicates a generic failure in the decoding of a MessageSet from the broker"""
+    pass
+
+
 class ProducerQueueFullError(KafkaException):
     """Indicates that one or more of the AsyncProducer's internal queues contain at least max_queued_messages messages"""
     pass

pykafka/protocol.py

@@ -65,7 +65,7 @@
 
 
 from .common import CompressionType, Message
-from .exceptions import ERROR_CODES, MessageSizeTooLarge
+from .exceptions import ERROR_CODES, MessageSetDecodeFailure
 from .utils import Serializable, compression, struct_helpers
 from .utils.compat import iteritems, itervalues, buffer
 
@@ -379,7 +379,7 @@ def decode(cls, buff, partition_id=-1):
             messages.append(message)
             offset += size
         if len(messages) == 0 and attempted:
-            raise MessageSizeTooLarge(size)
+            raise MessageSetDecodeFailure(size)
         return MessageSet(messages=messages)
 
     def pack_into(self, buff, offset):