New KafkaConfig Refactoring (#48)

* Refactor KafkaClient to use new Kafka configs

Motivation:

Currently, the new strongly-typed config structs such as ProducerConfig
are just used to create a new KafkaConfig. We want to get rid of
KafkaConfig entirely and only create librdkafka configs right before
rd_kafka_new() is called to avoid having to bother with the memory
management of a rd_kafka_conf_t object.

Modifications:

* deprecate KafkaConfig in favour of ProducerConfig, ConsumerConfig etc.

Result:

* everything should work as before, just with a slightly nicer syntax

* Temporarily remove socket.connection.setup.timeout.ms

Motivation:

* This option is only available with newer librdkafka versions, not
  including the librdkafka package used in our Docker image

Modifications:

* Temporarily commented out the option to pass unit tests on Linux

* Remove KafkaConfig

Modifications:

* remove KafkaConfig
* remove unused Tests
* update DocC to use ConsumerConfig and ProducerConfig

* Create internal wrapper for rd_kafka_conf funcs

Modifications:

* new struct `RDKafkaConfig` containing helper functions wrapping
  `rd_kafka_conf*` functions for easier use inside of `KafkaClient`

* Replace KafkaTopicConfig with new TopicConfig

Motivation:

* similar to Producer/Consumer configs, the librdkafka
  rd_kafka_topic_conf_t config shall only be created when needed by the
  rd_kafka_topic_new, for all other cases a light-weight Swift struct
  TopicConfig shall be used

Modifications:

* delete `KafkaTopicConfig` and `KafkaTopicConfigTests`
* create a new wrapper `RDKafkaTopicConfig` that wraps the common
  `rd_kafka_topic_conf_*` functions
* refactor `KafkaProducer` into using the new `TopicConfig` type

* Remove TODOs

* Rearrange folders + rename config structs

Motivation:

* create a clearer file structure
* add `Kafka*` prefix to configs to have a clear Kafka namespace and
  avoid name collisions when used in other projects

Modifications:

* created folders for Configuration, RDKafka and Utilities
* rename ConsumerConfig -> KafkaConsumerConfig
* rename ProducerConfig -> KafkaProducerConfig
* rename TopicConfig -> KafkaTopicConfig
* rename ConfigEnums -> KafkaConfigEnums

* RDKafkaConfig.setDeliveryCallback -> .setDeliveryReportCallback

* Update README to use new config types

* Remove unused DocC comment

* * re-enable socketConnectionSetupTimeoutMs configuration option in Kafka(Consumer|Producer)Config

* Review Franz

Modifications:

* rename `KafkaConfigEnums` to `KafkaSharedConfiguration`
* remove `protocol StringDictionaryRepresentable`

* KafkaProducer: ensure deliveryCallback is retained

Motivation:

With the current code architecture, `KafkaClient` has a var `opaque`
that references a `CapturedClosure` which in turn captures the delivery
callback that gets invoked by Kafka upon successful message delivery.
This callback has to be retained in memory while our producer is
running.

Modifications:

* make `KafkaClient` a `class` in which the captured closure is a `let`
  and not a `var`
* create a factory method called `RDKafka.createClient` that creates
  a new `KafkaClient` for a given set of parameters

Author: felixschlegel

README.md

@@ -9,8 +9,7 @@ SwiftKafka is a Swift Package in development that provides a convenient way to c
 The `sendAsync(_:)` method of `KafkaProducer` returns a message-id that can later be used to identify the corresponding acknowledgement. Acknowledgements are received through the `acknowledgements` [`AsyncSequence`](https://developer.apple.com/documentation/swift/asyncsequence). Each acknowledgement indicates that producing a message was successful or returns an error.
 
 ```swift
-var config = KafkaConfig()
-try config.set("localhost:9092", forKey: "bootstrap.servers")
+let config = KafkaProducerConfig(bootstrapServers: ["localhost:9092"])
 
 let producer = try await KafkaProducer(
     config: config,
@@ -37,8 +36,7 @@ await producer.shutdownGracefully()
 After initializing the `KafkaConsumer` with a topic-partition pair to read from, messages can be consumed using the `messages` [`AsyncSequence`](https://developer.apple.com/documentation/swift/asyncsequence).
 
 ```swift
-var config = KafkaConfig()
-try config.set("localhost:9092", forKey: "bootstrap.servers")
+let config = KafkaConsumerConfig(bootstrapServers: ["localhost:9092"])
 
 let consumer = try KafkaConsumer(
     topic: "topic-name",
@@ -62,12 +60,13 @@ for await messageResult in consumer.messages {
 SwiftKafka also allows users to subscribe to an array of topics as part of a consumer group.
 
 ```swift
-var config = KafkaConfig()
-try config.set("localhost:9092", forKey: "bootstrap.servers")
+let config = KafkaConsumerConfig(
+    groupID: "example-group",
+    bootstrapServers: ["localhost:9092"]
+)
 
 let consumer = try KafkaConsumer(
     topics: ["topic-name"],
-    groupID: "example-group",
     config: config,
     logger: .kafkaTest // Your logger here
 )
@@ -87,13 +86,14 @@ for await messageResult in consumer.messages {
 By default, the `KafkaConsumer` automatically commits message offsets after receiving the corresponding message. However, we allow users to disable this setting and commit message offsets manually.
 
 ```swift
-var config = KafkaConfig()
-try config.set("localhost:9092", forKey: "bootstrap.servers")
-try config.set("false", forKey: "enable.auto.commit")
+let config = KafkaConsumerConfig(
+    groupID: "example-group",
+    enableAutoCommit: false,
+    bootstrapServers: ["localhost:9092"]
+)
 
 let consumer = try KafkaConsumer(
     topics: ["topic-name"],
-    groupID: "example-group",
     config: config,
     logger: .kafkaTest // Your logger here
 )

Sources/SwiftKafka/New/ConsumerConfig.swift renamed to Sources/SwiftKafka/Configuration/KafkaConsumerConfig.swift

@@ -12,7 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
-public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable {
+public struct KafkaConsumerConfig: Hashable, Equatable {
     var dictionary: [String: String] = [:]
 
     // MARK: - Consumer-specific Config Properties
@@ -54,7 +54,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// Action to take when there is no initial offset in offset store or the desired offset is out of range. See ``ConfigEnums/AutoOffsetReset`` for more information.
-    public var autoOffsetReset: ConfigEnums.AutoOffsetReset {
+    public var autoOffsetReset: KafkaSharedConfiguration.AutoOffsetReset {
         get { self.getAutoOffsetReset() ?? .largest }
         set { self.dictionary["auto.offset.reset"] = newValue.description }
     }
@@ -146,7 +146,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// A comma-separated list of debug contexts to enable. Detailed Producer debugging: broker,topic,msg. Consumer: consumer,cgrp,topic,fetch.
-    public var debug: [ConfigEnums.DebugOption] {
+    public var debug: [KafkaSharedConfiguration.DebugOption] {
         get { self.getDebugOptions() }
         set {
             if !newValue.isEmpty {
@@ -204,7 +204,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// Allowed broker ``ConfigEnums/IPAddressFamily``.
-    public var brokerAddressFamily: ConfigEnums.IPAddressFamily {
+    public var brokerAddressFamily: KafkaSharedConfiguration.IPAddressFamily {
         get { self.getIPAddressFamily() ?? .any }
         set { self.dictionary["broker.address.family"] = newValue.description }
     }
@@ -222,7 +222,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// ``ConfigEnums/SecurityProtocol`` used to communicate with brokers.
-    public var securityProtocol: ConfigEnums.SecurityProtocol {
+    public var securityProtocol: KafkaSharedConfiguration.SecurityProtocol {
         get { self.getSecurityProtocol() ?? .plaintext }
         set { self.dictionary["security.protocol"] = newValue.description }
     }
@@ -270,7 +270,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// SASL mechanism to use for authentication.
-    public var saslMechanism: ConfigEnums.SASLMechanism? {
+    public var saslMechanism: KafkaSharedConfiguration.SASLMechanism? {
         get { self.getSASLMechanism() }
         set {
             if let newValue {
@@ -307,7 +307,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
         enableAutoCommit: Bool = true,
         autoCommitIntervalMs: UInt = 5000,
         enableAutoOffsetStore: Bool = true,
-        autoOffsetReset: ConfigEnums.AutoOffsetReset = .largest,
+        autoOffsetReset: KafkaSharedConfiguration.AutoOffsetReset = .largest,
         allowAutoCreateTopics: Bool = false,
         clientID: String = "rdkafka",
         bootstrapServers: [String] = [],
@@ -321,7 +321,7 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
         topicMetadataRefreshSparse: Bool = true,
         topicMetadataPropagationMaxMs: UInt = 30000,
         topicDenylist: [String] = [],
-        debug: [ConfigEnums.DebugOption] = [],
+        debug: [KafkaSharedConfiguration.DebugOption] = [],
         socketTimeoutMs: UInt = 60000,
         socketSendBufferBytes: UInt = 0,
         socketReceiveBufferBytes: UInt = 0,
@@ -330,18 +330,18 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
         socketMaxFails: UInt = 1,
         socketConnectionSetupTimeoutMs: UInt = 30000,
         brokerAddressTTL: UInt = 1000,
-        brokerAddressFamily: ConfigEnums.IPAddressFamily = .any,
+        brokerAddressFamily: KafkaSharedConfiguration.IPAddressFamily = .any,
         reconnectBackoffMs: UInt = 100,
         reconnectBackoffMaxMs: UInt = 10000,
-        securityProtocol: ConfigEnums.SecurityProtocol = .plaintext,
+        securityProtocol: KafkaSharedConfiguration.SecurityProtocol = .plaintext,
         sslKeyLocation: String = "",
         sslKeyPassword: String = "",
         sslCertificateLocation: String = "",
         sslCALocation: String = "",
         sslCRLLocation: String = "",
         sslKeystoreLocation: String = "",
         sslKeystorePassword: String = "",
-        saslMechanism: ConfigEnums.SASLMechanism? = nil,
+        saslMechanism: KafkaSharedConfiguration.SASLMechanism? = nil,
         saslUsername: String? = nil,
         saslPassword: String? = nil
     ) {
@@ -394,46 +394,46 @@ public struct ConsumerConfig: Hashable, Equatable, StringDictionaryRepresentable
 
     // MARK: - Helpers
 
-    func getDebugOptions() -> [ConfigEnums.DebugOption] {
+    func getDebugOptions() -> [KafkaSharedConfiguration.DebugOption] {
         guard let options = dictionary["debug"] else {
             return []
         }
         return options.components(separatedBy: ",")
-            .map { ConfigEnums.DebugOption(description: $0) }
+            .map { KafkaSharedConfiguration.DebugOption(description: $0) }
     }
 
-    func getIPAddressFamily() -> ConfigEnums.IPAddressFamily? {
+    func getIPAddressFamily() -> KafkaSharedConfiguration.IPAddressFamily? {
         guard let value = dictionary["broker.address.family"] else {
             return nil
         }
-        return ConfigEnums.IPAddressFamily(description: value)
+        return KafkaSharedConfiguration.IPAddressFamily(description: value)
     }
 
-    func getSecurityProtocol() -> ConfigEnums.SecurityProtocol? {
+    func getSecurityProtocol() -> KafkaSharedConfiguration.SecurityProtocol? {
         guard let value = dictionary["security.protocol"] else {
             return nil
         }
-        return ConfigEnums.SecurityProtocol(description: value)
+        return KafkaSharedConfiguration.SecurityProtocol(description: value)
     }
 
-    func getSASLMechanism() -> ConfigEnums.SASLMechanism? {
+    func getSASLMechanism() -> KafkaSharedConfiguration.SASLMechanism? {
         guard let value = dictionary["sasl.mechanism"] else {
             return nil
         }
-        return ConfigEnums.SASLMechanism(description: value)
+        return KafkaSharedConfiguration.SASLMechanism(description: value)
     }
 
-    func getAutoOffsetReset() -> ConfigEnums.AutoOffsetReset? {
+    func getAutoOffsetReset() -> KafkaSharedConfiguration.AutoOffsetReset? {
         guard let value = dictionary["auto.offset.reset"] else {
             return nil
         }
-        return ConfigEnums.AutoOffsetReset(description: value)
+        return KafkaSharedConfiguration.AutoOffsetReset(description: value)
     }
 }
 
 // MARK: - ConfigEnums + AutoOffsetReset
 
-extension ConfigEnums {
+extension KafkaSharedConfiguration {
     /// Available actions to take when there is no initial offset in offset store / offset is out of range.
     public struct AutoOffsetReset: Hashable, Equatable, CustomStringConvertible {
         public let description: String

Sources/SwiftKafka/New/ProducerConfig.swift renamed to Sources/SwiftKafka/Configuration/KafkaProducerConfig.swift

@@ -12,7 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
-public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable {
+public struct KafkaProducerConfig: Hashable, Equatable {
     var dictionary: [String: String] = [:]
 
     // MARK: - Producer-specific Config Properties
@@ -134,7 +134,7 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// A comma-separated list of debug contexts to enable. Detailed Producer debugging: broker,topic,msg. Consumer: consumer,cgrp,topic,fetch.
-    public var debug: [ConfigEnums.DebugOption] {
+    public var debug: [KafkaSharedConfiguration.DebugOption] {
         get { self.getDebugOptions() }
         set {
             if !newValue.isEmpty {
@@ -192,7 +192,7 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// Allowed broker ``ConfigEnums/IPAddressFamily``.
-    public var brokerAddressFamily: ConfigEnums.IPAddressFamily {
+    public var brokerAddressFamily: KafkaSharedConfiguration.IPAddressFamily {
         get { self.getIPAddressFamily() ?? .any }
         set { self.dictionary["broker.address.family"] = newValue.description }
     }
@@ -210,7 +210,7 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// ``ConfigEnums/SecurityProtocol`` used to communicate with brokers.
-    public var securityProtocol: ConfigEnums.SecurityProtocol {
+    public var securityProtocol: KafkaSharedConfiguration.SecurityProtocol {
         get { self.getSecurityProtocol() ?? .plaintext }
         set { self.dictionary["security.protocol"] = newValue.description }
     }
@@ -258,7 +258,7 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
     }
 
     /// SASL mechanism to use for authentication.
-    public var saslMechanism: ConfigEnums.SASLMechanism? {
+    public var saslMechanism: KafkaSharedConfiguration.SASLMechanism? {
         get { self.getSASLMechanism() }
         set {
             if let newValue {
@@ -307,7 +307,7 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
         topicMetadataRefreshSparse: Bool = true,
         topicMetadataPropagationMaxMs: UInt = 30000,
         topicDenylist: [String] = [],
-        debug: [ConfigEnums.DebugOption] = [],
+        debug: [KafkaSharedConfiguration.DebugOption] = [],
         socketTimeoutMs: UInt = 60000,
         socketSendBufferBytes: UInt = 0,
         socketReceiveBufferBytes: UInt = 0,
@@ -316,18 +316,18 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
         socketMaxFails: UInt = 1,
         socketConnectionSetupTimeoutMs: UInt = 30000,
         brokerAddressTTL: UInt = 1000,
-        brokerAddressFamily: ConfigEnums.IPAddressFamily = .any,
+        brokerAddressFamily: KafkaSharedConfiguration.IPAddressFamily = .any,
         reconnectBackoffMs: UInt = 100,
         reconnectBackoffMaxMs: UInt = 10000,
-        securityProtocol: ConfigEnums.SecurityProtocol = .plaintext,
+        securityProtocol: KafkaSharedConfiguration.SecurityProtocol = .plaintext,
         sslKeyLocation: String = "",
         sslKeyPassword: String = "",
         sslCertificateLocation: String = "",
         sslCALocation: String = "",
         sslCRLLocation: String = "",
         sslKeystoreLocation: String = "",
         sslKeystorePassword: String = "",
-        saslMechanism: ConfigEnums.SASLMechanism? = nil,
+        saslMechanism: KafkaSharedConfiguration.SASLMechanism? = nil,
         saslUsername: String? = nil,
         saslPassword: String? = nil
     ) {
@@ -377,32 +377,32 @@ public struct ProducerConfig: Hashable, Equatable, StringDictionaryRepresentable
 
     // MARK: - Helpers
 
-    func getDebugOptions() -> [ConfigEnums.DebugOption] {
+    func getDebugOptions() -> [KafkaSharedConfiguration.DebugOption] {
         guard let options = dictionary["debug"] else {
             return []
         }
         return options.components(separatedBy: ",")
-            .map { ConfigEnums.DebugOption(description: $0) }
+            .map { KafkaSharedConfiguration.DebugOption(description: $0) }
     }
 
-    func getIPAddressFamily() -> ConfigEnums.IPAddressFamily? {
+    func getIPAddressFamily() -> KafkaSharedConfiguration.IPAddressFamily? {
         guard let value = dictionary["broker.address.family"] else {
             return nil
         }
-        return ConfigEnums.IPAddressFamily(description: value)
+        return KafkaSharedConfiguration.IPAddressFamily(description: value)
     }
 
-    func getSecurityProtocol() -> ConfigEnums.SecurityProtocol? {
+    func getSecurityProtocol() -> KafkaSharedConfiguration.SecurityProtocol? {
         guard let value = dictionary["security.protocol"] else {
             return nil
         }
-        return ConfigEnums.SecurityProtocol(description: value)
+        return KafkaSharedConfiguration.SecurityProtocol(description: value)
     }
 
-    func getSASLMechanism() -> ConfigEnums.SASLMechanism? {
+    func getSASLMechanism() -> KafkaSharedConfiguration.SASLMechanism? {
         guard let value = dictionary["sasl.mechanism"] else {
             return nil
         }
-        return ConfigEnums.SASLMechanism(description: value)
+        return KafkaSharedConfiguration.SASLMechanism(description: value)
     }
 }

Sources/SwiftKafka/New/ConfigEnums.swift renamed to Sources/SwiftKafka/Configuration/KafkaSharedConfiguration.swift

@@ -13,7 +13,7 @@
 //===----------------------------------------------------------------------===//
 
 /// Collection of `enum` types used in the configuration structs this library provides.
-public struct ConfigEnums {
+public enum KafkaSharedConfiguration {
     /// Available debug contexts to enable.
     public struct DebugOption: Hashable, Equatable, CustomStringConvertible {
         public let description: String