Improved implementation of KafkaError and KafkaAcknowledgedMessageError (#36)

* Refactored KafkaError

Modifications:

* made KafkaError public
* distinction between rdKafkaError and anyError in KafkaError

* Implement more specific errors

* Refactored KafkaAcknowledgedMessageError

* Added DocC Throws comments + ran swiftformat

* Improved structure of KafkaError

Modifications:

* refactor KafkaError to have a forwared evolvable type that caputures
  as much information as possible (proposed by FranzBusch in #36)
* fixed two broken documentation links

* Refactor KafkaAcknowledgedMessageError

Modifications:

* change structure of KafkaAcknowledgedMessageError so that it only
  contains a messageID and a wrapped KafkaError
* introduced KafkaError.ErrorCode.acknowledgement

Author: felixschlegel

Sources/SwiftKafka/KafkaAcknowledgedMessage.swift

@@ -18,7 +18,7 @@ import NIOCore
 /// A message produced by the client and acknowledged by the Kafka cluster.
 public struct KafkaAcknowledgedMessage: Hashable {
     /// The unique identifier assigned by the ``KafkaProducer`` when the message was send to Kafka.
-    /// The same identifier is returned by ``KafkaProducer/sendAsync(message:)`` and can be used to correlate
+    /// The same identifier is returned by ``KafkaProducer/sendAsync(_:)`` and can be used to correlate
     /// a sent message and an acknowledged message.
     public var id: UInt
     /// The topic that the message was sent to.
@@ -33,6 +33,7 @@ public struct KafkaAcknowledgedMessage: Hashable {
     public var offset: Int64
 
     /// Initialize ``KafkaAcknowledgedMessage`` from `rd_kafka_message_t` pointer.
+    /// - Throws: A ``KafkaAcknowledgedMessageError`` for failed acknowledgements or malformed messages.
     init(messagePointer: UnsafePointer<rd_kafka_message_t>, id: UInt) throws {
         self.id = id
 
@@ -41,15 +42,15 @@ public struct KafkaAcknowledgedMessage: Hashable {
         let valueBufferPointer = UnsafeRawBufferPointer(start: rdKafkaMessage.payload, count: rdKafkaMessage.len)
         self.value = ByteBuffer(bytes: valueBufferPointer)
 
-        guard rdKafkaMessage.err.rawValue == 0 else {
+        guard rdKafkaMessage.err == RD_KAFKA_RESP_ERR_NO_ERROR else {
             var errorStringBuffer = self.value
             let errorString = errorStringBuffer.readString(length: errorStringBuffer.readableBytes)
 
-            throw KafkaAcknowledgedMessageError(
-                rawValue: rdKafkaMessage.err.rawValue,
-                description: errorString,
-                messageID: self.id
-            )
+            if let errorString {
+                throw KafkaAcknowledgedMessageError.fromMessage(messageID: self.id, message: errorString)
+            } else {
+                throw KafkaAcknowledgedMessageError.fromRDKafkaError(messageID: self.id, error: rdKafkaMessage.err)
+            }
         }
 
         guard let topic = String(validatingUTF8: rd_kafka_topic_name(rdKafkaMessage.rkt)) else {

Sources/SwiftKafka/KafkaAcknowledgedMessageError.swift

@@ -12,12 +12,53 @@
 //
 //===----------------------------------------------------------------------===//
 
+import Crdkafka
+
 /// Error caused by the Kafka cluster when trying to process a message produced by ``KafkaProducer``.
-public struct KafkaAcknowledgedMessageError: Error {
-    /// A raw value representing the error code.
-    public var rawValue: Int32
-    /// A string describing the error.
-    public var description: String?
+public struct KafkaAcknowledgedMessageError: Error, CustomStringConvertible {
     /// Identifier of the message that caused the error.
     public var messageID: UInt
+    /// The underlying ``KafkaError``.
+    public let error: KafkaError
+
+    init(messageID: UInt, error: KafkaError) {
+        self.messageID = messageID
+        self.error = error
+    }
+
+    public var description: String {
+        self.error.description
+    }
+
+    static func fromRDKafkaError(
+        messageID: UInt,
+        error: rd_kafka_resp_err_t,
+        file: String = #fileID,
+        line: UInt = #line
+    ) -> Self {
+        .init(
+            messageID: messageID,
+            error: .rdKafkaError(
+                wrapping: error,
+                file: file,
+                line: line
+            )
+        )
+    }
+
+    static func fromMessage(
+        messageID: UInt,
+        message: String,
+        file: String = #fileID,
+        line: UInt = #line
+    ) -> Self {
+        .init(
+            messageID: messageID,
+            error: .acknowledgement(
+                reason: message,
+                file: file,
+                line: line
+            )
+        )
+    }
 }

Sources/SwiftKafka/KafkaClient.swift

@@ -57,7 +57,7 @@ final class KafkaClient {
                 rd_kafka_conf_destroy(duplicateConfig)
 
                 let errorString = String(cString: errorChars)
-                throw KafkaError(description: errorString)
+                throw KafkaError.client(reason: errorString)
             }
 
             return handle

Sources/SwiftKafka/KafkaConfig.swift

@@ -90,7 +90,7 @@ public struct KafkaConfig: Hashable, Equatable {
 
             if configResult != RD_KAFKA_CONF_OK {
                 let errorString = String(cString: errorChars)
-                throw KafkaError(description: errorString)
+                throw KafkaError.config(reason: errorString)
             }
         }
 
@@ -166,6 +166,7 @@ public struct KafkaConfig: Hashable, Equatable {
     }
 
     /// Set configuration `value` for `key`
+    /// - Throws: A ``KafkaError`` if setting the value failed.
     public mutating func set(_ value: String, forKey key: String) throws {
         // Copy-on-write mechanism
         if !isKnownUniquelyReferenced(&(self._internal)) {

Sources/SwiftKafka/KafkaConsumer.swift

@@ -34,7 +34,7 @@ private struct ConsumerMessagesAsyncSequenceDelegate: NIOAsyncSequenceProducerDe
 
 /// `AsyncSequence` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
 public struct ConsumerMessagesAsyncSequence: AsyncSequence {
-    public typealias Element = Result<KafkaConsumerMessage, Error> // TODO: replace with something like KafkaConsumerError
+    public typealias Element = Result<KafkaConsumerMessage, KafkaError>
     typealias HighLowWatermark = NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark
     fileprivate let wrappedSequence: NIOAsyncSequenceProducer<Element, HighLowWatermark, ConsumerMessagesAsyncSequenceDelegate>
 
@@ -74,7 +74,7 @@ public final class KafkaConsumer {
 
     // We use implicitly unwrapped optionals here as these properties need to access self upon initialization
     /// Type of the values returned by the ``messages`` sequence.
-    private typealias Element = Result<KafkaConsumerMessage, Error> // TODO: replace with a more specific Error type
+    private typealias Element = Result<KafkaConsumerMessage, KafkaError>
     private var messagesSource: NIOAsyncSequenceProducer<
         Element,
         ConsumerMessagesAsyncSequence.HighLowWatermark,
@@ -88,6 +88,7 @@ public final class KafkaConsumer {
     /// or assign the consumer to a particular topic + partition pair using ``assign(topic:partition:offset:)``.
     /// - Parameter config: The ``KafkaConfig`` for configuring the ``KafkaConsumer``.
     /// - Parameter logger: A logger.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
     private init(
         config: KafkaConfig,
         logger: Logger
@@ -104,7 +105,7 @@ public final class KafkaConsumer {
             rd_kafka_poll_set_consumer(handle)
         }
         guard result == RD_KAFKA_RESP_ERR_NO_ERROR else {
-            throw KafkaError(rawValue: result.rawValue)
+            throw KafkaError.rdKafkaError(wrapping: result)
         }
 
         self.serialQueue = DispatchQueue(label: "swift-kafka-gsoc.consumer.serial")
@@ -136,6 +137,7 @@ public final class KafkaConsumer {
     /// - Parameter groupID: Name of the consumer group that this ``KafkaConsumer`` will create / join.
     /// - Parameter config: The ``KafkaConfig`` for configuring the ``KafkaConsumer``.
     /// - Parameter logger: A logger.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
     public convenience init(
         topics: [String],
         groupID: String,
@@ -145,7 +147,7 @@ public final class KafkaConsumer {
         var config = config
         if let configGroupID = config.value(forKey: "group.id") {
             if configGroupID != groupID {
-                throw KafkaError(description: "Group ID does not match with group ID found in the configuration")
+                throw KafkaError.config(reason: "Group ID does not match with group ID found in the configuration")
             }
         } else {
             try config.set(groupID, forKey: "group.id")
@@ -164,6 +166,7 @@ public final class KafkaConsumer {
     /// - Parameter offset: The topic offset where reading begins. Defaults to the offset of the last read message.
     /// - Parameter config: The ``KafkaConfig`` for configuring the ``KafkaConsumer``.
     /// - Parameter logger: A logger.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
     /// - Note: This consumer ignores the `group.id` property of its `config`.
     public convenience init(
         topic: String,
@@ -193,6 +196,7 @@ public final class KafkaConsumer {
     /// Subscribe to the given list of `topics`.
     /// The partition assignment happens automatically using `KafkaConsumer`'s consumer group.
     /// - Parameter topics: An array of topic names to subscribe to.
+    /// - Throws: A ``KafkaError`` if subscribing to the topic list failed.
     private func subscribe(topics: [String]) throws {
         assert(!closed)
 
@@ -208,15 +212,16 @@ public final class KafkaConsumer {
             rd_kafka_subscribe(handle, subscribedTopicsPointer)
         }
 
-        guard result.rawValue == 0 else {
-            throw KafkaError(rawValue: result.rawValue)
+        guard result == RD_KAFKA_RESP_ERR_NO_ERROR else {
+            throw KafkaError.rdKafkaError(wrapping: result)
         }
     }
 
     /// Assign the``KafkaConsumer`` to a specific `partition` of a `topic`.
     /// - Parameter topic: Name of the topic that this ``KafkaConsumer`` will read from.
     /// - Parameter partition: Partition that this ``KafkaConsumer`` will read from.
     /// - Parameter offset: The topic offset where reading begins. Defaults to the offset of the last read message.
+    /// - Throws: A ``KafkaError`` if the consumer could not be assigned to the topic + partition pair.
     private func assign(
         topic: String,
         partition: KafkaPartition,
@@ -239,7 +244,7 @@ public final class KafkaConsumer {
         }
 
         guard result == RD_KAFKA_RESP_ERR_NO_ERROR else {
-            throw KafkaError(rawValue: result.rawValue)
+            throw KafkaError.rdKafkaError(wrapping: result)
         }
     }
 
@@ -257,8 +262,11 @@ public final class KafkaConsumer {
                     return
                 }
                 messageResult = .success(message)
+            } catch let kafkaError as KafkaError {
+                messageResult = .failure(kafkaError)
             } catch {
-                messageResult = .failure(error)
+                self.logger.error("KafkaConsumer caught error: \(error)")
+                return
             }
 
             let yieldresult = self.messagesSource.yield(messageResult)
@@ -275,6 +283,7 @@ public final class KafkaConsumer {
     /// This method blocks for a maximum of `timeout` milliseconds.
     /// - Parameter timeout: Maximum amount of milliseconds this method waits for a new message.
     /// - Returns: A ``KafkaConsumerMessage`` or `nil` if there are no new messages.
+    /// - Throws: A ``KafkaError`` if the received message is an error message or malformed.
     private func poll(timeout: Int32 = 100) throws -> KafkaConsumerMessage? {
         dispatchPrecondition(condition: .onQueue(self.serialQueue))
         assert(!closed)
@@ -307,6 +316,7 @@ public final class KafkaConsumer {
     /// Mark `message` in the topic as read and request the next message from the topic.
     /// This method is only used for manual offset management.
     /// - Parameter message: Last received message that shall be marked as read.
+    /// - Throws: A ``KafkaError`` if committing failed.
     /// - Warning: This method fails if the `enable.auto.commit` configuration property is set to `true`.
     public func commitSync(_ message: KafkaConsumerMessage) async throws {
         try await self.serializeWithThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
@@ -323,11 +333,11 @@ public final class KafkaConsumer {
     private func _commitSync(_ message: KafkaConsumerMessage) throws {
         dispatchPrecondition(condition: .onQueue(self.serialQueue))
         guard !self.closed else {
-            throw KafkaError(description: "Trying to invoke method on consumer that has been closed.")
+            throw KafkaError.connectionClosed(reason: "Tried to commit message offset on a closed consumer")
         }
 
         guard self.config.value(forKey: "enable.auto.commit") == "false" else {
-            throw KafkaError(description: "Committing manually only works if enable.auto.commit is set to false")
+            throw KafkaError.config(reason: "Committing manually only works if enable.auto.commit is set to false")
         }
 
         let changesList = rd_kafka_topic_partition_list_new(1)
@@ -352,7 +362,7 @@ public final class KafkaConsumer {
             )
         }
         guard result == RD_KAFKA_RESP_ERR_NO_ERROR else {
-            throw KafkaError(rawValue: result.rawValue)
+            throw KafkaError.rdKafkaError(wrapping: result)
         }
         return
     }
@@ -371,7 +381,7 @@ public final class KafkaConsumer {
             rd_kafka_topic_partition_list_destroy(self.subscribedTopicsPointer)
 
             guard result == RD_KAFKA_RESP_ERR_NO_ERROR else {
-                let error = KafkaError(rawValue: result.rawValue)
+                let error = KafkaError.rdKafkaError(wrapping: result)
                 self.logger.error("Closing KafkaConsumer failed: \(error.description)")
                 return
             }

Sources/SwiftKafka/KafkaConsumerMessage.swift

@@ -29,6 +29,7 @@ public struct KafkaConsumerMessage: Hashable {
     public var offset: Int64
 
     /// Initialize ``KafkaConsumerMessage`` from `rd_kafka_message_t` pointer.
+    /// - Throws: A ``KafkaError`` if the received message is an error message or malformed.
     init(messagePointer: UnsafePointer<rd_kafka_message_t>) throws {
         let rdKafkaMessage = messagePointer.pointee
 
@@ -42,10 +43,11 @@ public struct KafkaConsumerMessage: Hashable {
             var errorStringBuffer = ByteBuffer(bytes: valueBufferPointer)
             let errorString = errorStringBuffer.readString(length: errorStringBuffer.readableBytes)
 
-            throw KafkaError(
-                rawValue: rdKafkaMessage.err.rawValue,
-                description: errorString ?? ""
-            )
+            if let errorString {
+                throw KafkaError.messageConsumption(reason: errorString)
+            } else {
+                throw KafkaError.rdKafkaError(wrapping: rdKafkaMessage.err)
+            }
         }
 
         guard let topic = String(validatingUTF8: rd_kafka_topic_name(rdKafkaMessage.rkt)) else {