Move KafkaClient to RD* namespace (#89)

Motivation:

Our `KafkaClient` class has evolved into a wrapper of `rd_kafka_*` C
functions. Therefore I thought it is reasonable to move it to the
`RDKafka*` namespace (conceptually) and remove the imports of `CRDKafka`
in `KafkaConsumer` and `KafkaProducer`.

Modifications:

* `KafkaConsumer`: remove `CRDKafka` import
* `KafkaProducer`: remove `CRDKafka` import
* rename `KafkaClient -> `RDKafkaClient` and move it to the `RDKafka`
  folder
* delete `RDKafka.swift` and move `create/makeClient` factory method to
  `RDKafkaClient`

Author: felixschlegel

Sources/SwiftKafka/KafkaConsumer.swift

@@ -12,7 +12,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-import Crdkafka
 import Logging
 import NIOConcurrencyHelpers
 import NIOCore
@@ -122,7 +121,7 @@ public final class KafkaConsumer: Sendable, Service {
         self.config = config
         self.logger = logger
 
-        let client = try RDKafka.createClient(
+        let client = try RDKafkaClient.makeClient(
             type: .consumer,
             configDictionary: config.dictionary,
             events: [.log, .fetch, .offsetCommit],
@@ -149,8 +148,8 @@ public final class KafkaConsumer: Sendable, Service {
             )
         }
 
-        // Events that would be triggered by ``KafkaClient/poll(timeout:)``
-        // are now triggered by ``KafkaClient/consumerPoll``.
+        // Events that would be triggered by ``RDKafkaClient/poll(timeout:)``
+        // are now triggered by ``RDKafkaClient/consumerPoll``.
         try client.pollSetConsumer()
 
         switch config.consumptionStrategy._internal {
@@ -289,7 +288,7 @@ public final class KafkaConsumer: Sendable, Service {
     }
 
     private func _triggerGracefulShutdown(
-        client: KafkaClient,
+        client: RDKafkaClient,
         logger: Logger
     ) {
         do {
@@ -323,27 +322,27 @@ extension KafkaConsumer {
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case initializing(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source
             )
             /// The ``KafkaConsumer`` is consuming messages.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case consuming(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source
             )
             /// Consumer is still running but the messages asynchronous sequence was terminated.
             /// All incoming messages will be dropped.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case consumptionStopped(client: KafkaClient)
+            case consumptionStopped(client: RDKafkaClient)
             /// The ``KafkaConsumer/triggerGracefulShutdown()`` has been invoked.
             /// We are now in the process of commiting our last state to the broker.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case finishing(client: KafkaClient)
+            case finishing(client: RDKafkaClient)
             /// The ``KafkaConsumer`` is closed.
             case finished
         }
@@ -354,7 +353,7 @@ extension KafkaConsumer {
         /// Delayed initialization of `StateMachine` as the `source` and the `pollClosure` are
         /// not yet available when the normal initialization occurs.
         mutating func initialize(
-            client: KafkaClient,
+            client: RDKafkaClient,
             source: Producer.Source
         ) {
             guard case .uninitialized = self.state else {
@@ -373,15 +372,15 @@ extension KafkaConsumer {
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case pollForAndYieldMessage(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source
             )
             /// The ``KafkaConsumer`` stopped consuming messages or
             /// is in the process of shutting down.
             /// Poll to serve any queued events and commit outstanding state to the broker.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case pollWithoutYield(client: KafkaClient)
+            case pollWithoutYield(client: RDKafkaClient)
             /// Terminate the poll loop.
             case terminatePollLoop
         }
@@ -416,7 +415,7 @@ extension KafkaConsumer {
         enum SetUpConnectionAction {
             /// Set up the connection through ``subscribe()`` or ``assign()``.
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case setUpConnection(client: KafkaClient)
+            case setUpConnection(client: RDKafkaClient)
         }
 
         /// Get action to be taken when wanting to set up the connection through ``subscribe()`` or ``assign()``.
@@ -458,7 +457,7 @@ extension KafkaConsumer {
         enum StoreOffsetAction {
             /// Store the message offset with the given `client`.
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case storeOffset(client: KafkaClient)
+            case storeOffset(client: RDKafkaClient)
         }
 
         /// Get action to take when wanting to store a message offset (to be auto-committed by `librdkafka`).
@@ -483,7 +482,7 @@ extension KafkaConsumer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             case commitSync(
-                client: KafkaClient
+                client: RDKafkaClient
             )
             /// Throw an error. The ``KafkaConsumer`` is closed.
             case throwClosedError
@@ -514,14 +513,14 @@ extension KafkaConsumer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             case triggerGracefulShutdown(
-                client: KafkaClient
+                client: RDKafkaClient
             )
             /// Shut down the ``KafkaConsumer`` and finish the given `source` object.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case triggerGracefulShutdownAndFinishSource(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source
             )
         }

Sources/SwiftKafka/KafkaProducer.swift

@@ -12,7 +12,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-import Crdkafka
 import Logging
 import NIOConcurrencyHelpers
 import NIOCore
@@ -119,7 +118,7 @@ public final class KafkaProducer: Service, Sendable {
     ) throws -> KafkaProducer {
         let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
 
-        let client = try RDKafka.createClient(
+        let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: config.dictionary,
             events: [.log], // No .deliveryReport here!
@@ -169,7 +168,7 @@ public final class KafkaProducer: Service, Sendable {
         )
         let source = sourceAndSequence.source
 
-        let client = try RDKafka.createClient(
+        let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: config.dictionary,
             events: [.log, .deliveryReport],
@@ -283,7 +282,7 @@ extension KafkaProducer {
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             /// - Parameter topicHandles: Class containing all topic names with their respective `rd_kafka_topic_t` pointer.
             case started(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 messageIDCounter: UInt,
                 source: Producer.Source?,
                 topicHandles: RDKafkaTopicHandles
@@ -292,14 +291,14 @@ extension KafkaProducer {
             /// All incoming acknowledgements will be dropped.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case consumptionStopped(client: KafkaClient)
+            case consumptionStopped(client: RDKafkaClient)
             /// ``KafkaProducer/triggerGracefulShutdown()`` was invoked so we are flushing
             /// any messages that wait to be sent and serve any remaining queued callbacks.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case flushing(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source?
             )
             /// The ``KafkaProducer`` has been shut down and cannot be used anymore.
@@ -312,7 +311,7 @@ extension KafkaProducer {
         /// Delayed initialization of `StateMachine` as the `source` is not yet available
         /// when the normal initialization occurs.
         mutating func initialize(
-            client: KafkaClient,
+            client: RDKafkaClient,
             source: Producer.Source?
         ) {
             guard case .uninitialized = self.state else {
@@ -331,12 +330,12 @@ extension KafkaProducer {
             /// Poll client.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case pollWithoutYield(client: KafkaClient)
+            case pollWithoutYield(client: RDKafkaClient)
             /// Poll client and yield acknowledgments if any received.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
-            case pollAndYield(client: KafkaClient, source: Producer.Source?)
+            case pollAndYield(client: RDKafkaClient, source: Producer.Source?)
             /// Terminate the poll loop and finish the given `NIOAsyncSequenceProducerSource`.
             ///
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
@@ -375,7 +374,7 @@ extension KafkaProducer {
             ///
             /// - Important: `newMessageID` is the new message ID assigned to the message to be sent.
             case send(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 newMessageID: UInt,
                 topicHandles: RDKafkaTopicHandles
             )

Sources/SwiftKafka/RDKafka/RDKafka.swift

@@ -17,7 +17,13 @@ import Logging
 
 /// Base class for ``KafkaProducer`` and ``KafkaConsumer``,
 /// which is used to handle the connection to the Kafka ecosystem.
-final class KafkaClient: Sendable {
+final class RDKafkaClient: Sendable {
+    /// Determines if client is a producer or a consumer.
+    enum ClientType {
+        case producer
+        case consumer
+    }
+
     // Default size for Strings returned from C API
     static let stringSize = 1024
 
@@ -29,7 +35,8 @@ final class KafkaClient: Sendable {
     /// `librdkafka`'s main `rd_kafka_queue_t`.
     private let mainQueue: OpaquePointer
 
-    init(
+    // Use factory method to initialize
+    private init(
         kafkaHandle: OpaquePointer,
         logger: Logger
     ) {
@@ -45,6 +52,44 @@ final class KafkaClient: Sendable {
         rd_kafka_destroy(kafkaHandle)
     }
 
+    /// Factory method creating a new instance of a ``RDKafkaClient``.
+    static func makeClient(
+        type: ClientType,
+        configDictionary: [String: String],
+        events: [RDKafkaEvent],
+        logger: Logger
+    ) throws -> RDKafkaClient {
+        let clientType = type == .producer ? RD_KAFKA_PRODUCER : RD_KAFKA_CONSUMER
+
+        let rdConfig = try RDKafkaConfig.createFrom(configDictionary: configDictionary)
+        // Manually override some of the configuration options
+        // Handle logs in event queue
+        try RDKafkaConfig.set(configPointer: rdConfig, key: "log.queue", value: "true")
+        // KafkaConsumer is manually storing read offsets
+        if type == .consumer {
+            try RDKafkaConfig.set(configPointer: rdConfig, key: "enable.auto.offset.store", value: "false")
+        }
+        RDKafkaConfig.setEvents(configPointer: rdConfig, events: events)
+
+        let errorChars = UnsafeMutablePointer<CChar>.allocate(capacity: RDKafkaClient.stringSize)
+        defer { errorChars.deallocate() }
+
+        guard let handle = rd_kafka_new(
+            clientType,
+            rdConfig,
+            errorChars,
+            RDKafkaClient.stringSize
+        ) else {
+            // rd_kafka_new only frees the rd_kafka_conf_t upon success
+            rd_kafka_conf_destroy(rdConfig)
+
+            let errorString = String(cString: errorChars)
+            throw KafkaError.client(reason: errorString)
+        }
+
+        return RDKafkaClient(kafkaHandle: handle, logger: logger)
+    }
+
     /// Produce a message to the Kafka cluster.
     ///
     /// - Parameter message: The ``KafkaProducerMessage`` that is sent to the KafkaCluster.
@@ -134,7 +179,7 @@ final class KafkaClient: Sendable {
     /// Handle event of type `RDKafkaEvent.deliveryReport`.
     ///
     /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.
-    /// - Returns: `KafkaEvent` to be returned as part of ``KafkaClient.eventPoll()`.
+    /// - Returns: `KafkaEvent` to be returned as part of ``RDKafkaClient.eventPoll()`.
     private func handleDeliveryReportEvent(_ event: OpaquePointer?) -> KafkaEvent {
         let deliveryReportCount = rd_kafka_event_message_count(event)
         var deliveryReportResults = [Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>]()
@@ -230,13 +275,13 @@ final class KafkaClient: Sendable {
         actualCallback(.success(()))
     }
 
-    /// Redirect the main ``KafkaClient/poll(timeout:)`` queue to the `KafkaConsumer`'s
-    /// queue (``KafkaClient/consumerPoll``).
+    /// Redirect the main ``RDKafkaClient/poll(timeout:)`` queue to the `KafkaConsumer`'s
+    /// queue (``RDKafkaClient/consumerPoll``).
     ///
-    /// Events that would be triggered by ``KafkaClient/poll(timeout:)``
-    /// are now triggered by ``KafkaClient/consumerPoll``.
+    /// Events that would be triggered by ``RDKafkaClient/poll(timeout:)``
+    /// are now triggered by ``RDKafkaClient/consumerPoll``.
     ///
-    /// - Warning: It is not allowed to call ``KafkaClient/poll(timeout:)`` after ``KafkaClient/pollSetConsumer``.
+    /// - Warning: It is not allowed to call ``RDKafkaClient/poll(timeout:)`` after ``RDKafkaClient/pollSetConsumer``.
     func pollSetConsumer() throws {
         let result = rd_kafka_poll_set_consumer(self.kafkaHandle)
         if result != RD_KAFKA_RESP_ERR_NO_ERROR {
@@ -347,7 +392,7 @@ final class KafkaClient: Sendable {
     /// Close the consumer asynchronously. This means revoking its assignemnt, committing offsets to broker and
     /// leaving the consumer group (if applicable).
     ///
-    /// Make sure to run poll loop until ``KafkaClient/consumerIsClosed`` returns `true`.
+    /// Make sure to run poll loop until ``RDKafkaClient/consumerIsClosed`` returns `true`.
     func consumerClose() throws {
         let consumerQueue = rd_kafka_queue_get_consumer(self.kafkaHandle)
         let result = rd_kafka_consumer_close_queue(self.kafkaHandle, consumerQueue)