Make KafkaError instantiable

Author: rnpridgeon

confluent_kafka/__init__.py

@@ -18,11 +18,10 @@
 
 from .deserializing_consumer import DeserializingConsumer
 from .serializing_producer import SerializingProducer
+from .error import KafkaException, KafkaError
 
 from .cimpl import (Producer,
                     Consumer,
-                    KafkaError,
-                    KafkaException,
                     Message,
                     TopicPartition,
                     libversion,

confluent_kafka/deserializing_consumer.py

@@ -16,11 +16,11 @@
 # limitations under the License.
 #
 
-from confluent_kafka.cimpl import (KafkaError,
-                                   Consumer as _ConsumerImpl)
-from .error import ConsumeError
-from .serialization import (SerializationError,
-                            SerializationContext,
+from confluent_kafka.cimpl import Consumer as _ConsumerImpl
+from .error import (ConsumeError,
+                    KeyDeserializationError,
+                    ValueDeserializationError)
+from .serialization import (SerializationContext,
                             MessageField)
 
 
@@ -115,6 +115,12 @@ def poll(self, timeout=-1):
             :py:class:`Message` or None on timeout
 
         Raises:
+            KeyDeserializationError: If an error occurs during key
+            deserialization.
+
+            ValueDeserializationError: If an error occurs during value
+            deserialization.
+
             ConsumeError if an error was encountered while polling.
 
         """
@@ -131,20 +137,16 @@ def poll(self, timeout=-1):
         if self._value_deserializer is not None:
             try:
                 value = self._value_deserializer(value, ctx)
-            except SerializationError as se:
-                raise ConsumeError(KafkaError._VALUE_DESERIALIZATION,
-                                   reason=se.message,
-                                   message=msg)
+            except Exception as se:
+                raise ValueDeserializationError(exception=se, message=msg)
 
         key = msg.key()
         ctx.field = MessageField.KEY
         if self._key_deserializer is not None:
             try:
                 key = self._key_deserializer(key, ctx)
-            except SerializationError as se:
-                raise ConsumeError(KafkaError._KEY_DESERIALIZATION,
-                                   reason=se.message,
-                                   message=msg)
+            except Exception as se:
+                raise KeyDeserializationError(exception=se, message=msg)
 
         msg.set_key(key)
         msg.set_value(value)

confluent_kafka/error.py

@@ -15,11 +15,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+from confluent_kafka.cimpl import KafkaException, KafkaError
 
+from confluent_kafka.serialization import SerializationError
 
-class ConsumeError(Exception):
-    __slots__ = ['message', 'reason', 'error_code']
 
+class ConsumeError(KafkaException):
     """
     Wraps all errors encountered during the consumption of a message.
 
@@ -28,25 +29,113 @@ class ConsumeError(Exception):
         may be retrieved from the ``message`` attribute.
 
     Args:
-        error (KafkaError): The error that occurred.
+        error_code (KafkaError): Error code indicating the type of error.
 
-        message (Message, optional): The message returned from the broker.
+        exception(Exception, optional): The original exception
 
-        reason (str): String description of the error.
+        message (Message, optional): The Kafka Message returned from the broker.
 
     """
+    def __init__(self, error_code, exception=None, message=None):
+        if exception is not None:
+            kafka_error = KafkaError(error_code, repr(exception))
+            self.exception = exception
+        else:
+            kafka_error = KafkaError(error_code)
+            self.exception = None
 
-    def __init__(self, error, reason=None, message=None):
-        self.error = error
-        if reason is None:
-            reason = error.str()
-
-        self.reason = reason
+        super(ConsumeError, self).__init__(kafka_error)
         self.message = message
 
-    def __repr__(self):
-        return str(self)
+    @property
+    def code(self):
+        return self.code()
+
+    @property
+    def name(self):
+        return self.name()
+
+
+class KeyDeserializationError(ConsumeError, SerializationError):
+    """
+    Wraps all errors encountered during the deserialization of a Kafka
+    Message's key.
+
+    Args:
+        exception(Exception, optional): The original exception
+
+        message (Message, optional): The Kafka Message returned from the broker.
+
+    """
+    def __init__(self, exception=None, message=None):
+        super(KeyDeserializationError, self).__init__(
+            KafkaError._KEY_DESERIALIZATION, exception=exception, message=message)
+
+
+class ValueDeserializationError(ConsumeError, SerializationError):
+    """
+    Wraps all errors encountered during the deserialization of a Kafka
+    Message's value.
+
+    Args:
+        exception(Exception, optional): The original exception
+
+        message (Message, optional): The Kafka Message returned from the broker.
+
+    """
+    def __init__(self, exception=None, message=None):
+        super(ValueDeserializationError, self).__init__(
+            KafkaError._VALUE_DESERIALIZATION, exception=exception, message=message)
+
+
+class ProduceError(KafkaException):
+    """
+    Wraps all errors encountered when Producing messages.
+
+    Args:
+        error_code (KafkaError): Error code indicating the type of error.
+
+        exception(Exception, optional): The original exception.
+
+    """
+    def __init__(self, error_code, exception=None):
+        if exception is not None:
+            kafka_error = KafkaError(error_code, repr(exception))
+            self.exception = exception
+        else:
+            kafka_error = KafkaError(error_code)
+            self.exception = None
+
+        super(ProduceError, self).__init__(kafka_error)
+
+    @property
+    def code(self):
+        return self.code()
+
+    @property
+    def name(self):
+        return self.name()
+
 
-    def __str__(self):
-        return "{} (KafkaError code {})".format(self.reason,
-                                                self.error)
+class KeySerializationError(ProduceError, SerializationError):
+    """
+    Wraps all errors encountered during the serialization of a Message key.
+
+    Args:
+        exception (Exception): The exception that occurred during serialization.
+    """
+    def __init__(self, exception=None):
+        super(KeySerializationError, self).__init__(
+            KafkaError._KEY_SERIALIZATION, exception=exception)
+
+
+class ValueSerializationError(ProduceError, SerializationError):
+    """
+    Wraps all errors encountered during the serialization of a Message value.
+
+    Args:
+        exception (Exception): The exception that occurred during serialization.
+    """
+    def __init__(self, exception=None):
+        super(ValueSerializationError, self).__init__(
+            KafkaError._VALUE_SERIALIZATION, exception=exception)

confluent_kafka/serialization/__init__.py

@@ -16,6 +16,7 @@
 # limitations under the License.
 #
 import struct as _struct
+from confluent_kafka.error import KafkaException
 
 __all__ = ['Deserializer',
            'IntegerDeserializer',
@@ -61,35 +62,8 @@ def __init__(self, topic, field):
         self.field = field
 
 
-class SerializationError(Exception):
+class SerializationError(KafkaException):
     """Generic error from serializer package"""
-
-    def __init__(self, message):
-        self.message = message
-
-    def __repr__(self):
-        return '{klass}(error={error})'.format(
-            klass=self.__class__.__name__,
-            error=self.message
-        )
-
-    def __str__(self):
-        return self.message
-
-
-class KeySerializationError(SerializationError):
-    pass
-
-
-class KeyDeserializationError(SerializationError):
-    pass
-
-
-class ValueSerializationError(SerializationError):
-    pass
-
-
-class ValueDeserializationError(SerializationError):
     pass
 
 

confluent_kafka/serializing_producer.py

@@ -16,9 +16,11 @@
 # limitations under the License.
 #
 
-from .cimpl import Producer as _ProducerImpl
+from confluent_kafka.cimpl import Producer as _ProducerImpl
 from .serialization import (MessageField,
                             SerializationContext)
+from .error import (KeySerializationError,
+                    ValueSerializationError)
 
 
 class SerializingProducer(_ProducerImpl):
@@ -152,16 +154,26 @@ def produce(self, topic, key=None, value=None, partition=-1,
                 the application should call :py:func:`SerializingProducer.Poll`
                 and try again.
 
-             KafkaException: for other errors, see exception code
+            KeySerializationError: If an error occurs during key serialization.
+
+            ValueSerializationError: If an error occurs during value
+            serialization.
+
+             ProduceException: For all other errors
 
         """
         ctx = SerializationContext(topic, MessageField.KEY)
         if self._key_serializer is not None:
-            key = self._key_serializer(key, ctx)
-
+            try:
+                key = self._key_serializer(key, ctx)
+            except Exception as se:
+                raise KeySerializationError(se)
         ctx.field = MessageField.VALUE
         if self._value_serializer is not None:
-            value = self._value_serializer(value, ctx)
+            try:
+                value = self._value_serializer(value, ctx)
+            except Exception as se:
+                raise ValueSerializationError(se)
 
         super(SerializingProducer, self).produce(topic, value, key,
                                                  headers=headers,