KafkaConsumerConfig add SwiftKafka options (#54)

* `KafkaConsumerConfig` add `SwiftKafka` options

Motivation:

We want to use the `KafkaConsumerConfig` to determine if the
`KafkaConsumer` is part of a consumer-group (subscription) or assinged
to a single partition of a topic (assingment). This should also be
implemented in a way that errors are caught at compile time.
Furthermore, we want users to be able to determine a
`BackPressureStrategy` for their respective `KafkaClient`.

Modifications:

* add two new configuration options to `KafkaConsumerConfig`:
  `ConsumptionStrategy` and `BackPressureStrategy`
* remove `KafkaConsumerConfig`'s `groupID` option, this is now taken
  care of by the `consumptionStrategy` option, which is better as this
  ensures at compile time that now groupID mismatch errors occur
* rename `KafkaProducerConfig` back to `ConfigEnums` as it is not really
  meant for shared configurations, rather as a namespace for _all_
  configuration options
* update README

* * move setting group.id to `KafkaConsumerConfig`

* * fix soundness

* * rename `ConfigEnums` -> `KafkaSharedConfiguration`

* Review Franz

Modifications:

* fix wrong Docc comments
* remove implicitly unwrapped optional
* rename `highLowWatermark(lowWatermark:,highWatermark:)` to
  `watermark(low:,high:)`
* `ConsumptionStrategy`: remove *Based suffixes
* replace type of `KafkaConsumer.offset`: `Int64` -> `Int`

* * fix docc nits

Author: felixschlegel

README.md

@@ -36,11 +36,15 @@ 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
-let config = KafkaConsumerConfig(bootstrapServers: ["localhost:9092"])
+let config = KafkaConsumerConfig(
+    consumptionStrategy: .partition(
+        topic: "topic-name",
+        partition: KafkaPartition(rawValue: 0)
+    ),
+    bootstrapServers: ["localhost:9092"]
+)
 
 let consumer = try KafkaConsumer(
-    topic: "topic-name",
-    partition: KafkaPartition(rawValue: 0),
     config: config,
     logger: .kafkaTest // Your logger here
 )
@@ -61,12 +65,11 @@ SwiftKafka also allows users to subscribe to an array of topics as part of a con
 
 ```swift
 let config = KafkaConsumerConfig(
-    groupID: "example-group",
+    consumptionStrategy: .group(groupID: "example-group", topics: ["topic-name"]),
     bootstrapServers: ["localhost:9092"]
 )
 
 let consumer = try KafkaConsumer(
-    topics: ["topic-name"],
     config: config,
     logger: .kafkaTest // Your logger here
 )
@@ -87,13 +90,12 @@ By default, the `KafkaConsumer` automatically commits message offsets after rece
 
 ```swift
 let config = KafkaConsumerConfig(
-    groupID: "example-group",
+    consumptionStrategy: .group(groupID: "example-group", topics: ["topic-name"]),
     enableAutoCommit: false,
     bootstrapServers: ["localhost:9092"]
 )
 
 let consumer = try KafkaConsumer(
-    topics: ["topic-name"],
     config: config,
     logger: .kafkaTest // Your logger here
 )

Sources/SwiftKafka/Configuration/KafkaConsumerConfig.swift

@@ -12,17 +12,50 @@
 //
 //===----------------------------------------------------------------------===//
 
+import Crdkafka
+import struct Foundation.UUID
+
 public struct KafkaConsumerConfig: Hashable, Equatable {
-    var dictionary: [String: String] = [:]
+    // MARK: - SwiftKafka-specific Config properties
 
-    // MARK: - Consumer-specific Config Properties
+    /// The backpressure strategy to be used for message consumption.
+    public var backPressureStrategy: KafkaSharedConfiguration.BackPressureStrategy = .watermark(
+        low: 10,
+        high: 50
+    )
+
+    // This backs the consumptionStrategy computed property.
+    private var _consumptionStrategy: KafkaSharedConfiguration.ConsumptionStrategy
 
-    /// Client group id string. All clients sharing the same group.id belong to the same group.
-    public var groupID: String {
-        get { self.dictionary["group.id"] ?? "" }
-        set { self.dictionary["group.id"] = String(newValue) }
+    /// The strategy used for consuming messages.
+    /// See ``KafkaSharedConfiguration/ConsumptionStrategy`` for more information.
+    public var consumptionStrategy: KafkaSharedConfiguration.ConsumptionStrategy {
+        get { self._consumptionStrategy }
+        set {
+            self._consumptionStrategy = newValue
+
+            // We do not expose the group.id option to the user
+            // but rather set it ourselves as part of our much safer
+            // consumptionStrategy option.
+            switch newValue._internal {
+            case .partition:
+                // Although an assignment is not related to a consumer group,
+                // librdkafka requires us to set a `group.id`.
+                // This is a known issue:
+                // https://github.com/edenhill/librdkafka/issues/3261
+                self.dictionary["group.id"] = UUID().uuidString
+            case .group(groupID: let groupID, topics: _):
+                self.dictionary["group.id"] = groupID
+            }
+        }
     }
 
+    // MARK: - librdkafka Config properties
+
+    var dictionary: [String: String] = [:]
+
+    // MARK: - Consumer-specific Config Properties
+
     /// Client group session and failure detection timeout. The consumer sends periodic heartbeats (heartbeat.interval.ms) to indicate its liveness to the broker. If no hearts are received by the broker for a group member within the session timeout, the broker will remove the consumer from the group and trigger a rebalance. The allowed range is configured with the broker configuration properties group.min.session.timeout.ms and group.max.session.timeout.ms. Also see max.poll.interval.ms.
     public var sessionTimeoutMs: UInt {
         get { self.dictionary.getUInt("session.timeout.ms") ?? 45000 }
@@ -53,7 +86,7 @@ public struct KafkaConsumerConfig: Hashable, Equatable {
         set { self.dictionary["auto.commit.interval.ms"] = String(newValue) }
     }
 
-    /// 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.
+    /// Action to take when there is no initial offset in offset store or the desired offset is out of range. See ``KafkaSharedConfiguration/AutoOffsetReset`` for more information.
     public var autoOffsetReset: KafkaSharedConfiguration.AutoOffsetReset {
         get { self.getAutoOffsetReset() ?? .largest }
         set { self.dictionary["auto.offset.reset"] = newValue.description }
@@ -205,7 +238,7 @@ public struct KafkaConsumerConfig: Hashable, Equatable {
         set { self.dictionary["broker.address.ttl"] = String(newValue) }
     }
 
-    /// Allowed broker ``ConfigEnums/IPAddressFamily``.
+    /// Allowed broker ``KafkaSharedConfiguration/IPAddressFamily``.
     public var brokerAddressFamily: KafkaSharedConfiguration.IPAddressFamily {
         get { self.getIPAddressFamily() ?? .any }
         set { self.dictionary["broker.address.family"] = newValue.description }
@@ -223,7 +256,7 @@ public struct KafkaConsumerConfig: Hashable, Equatable {
         set { self.dictionary["reconnect.backoff.max.ms"] = String(newValue) }
     }
 
-    /// ``ConfigEnums/SecurityProtocol`` used to communicate with brokers.
+    /// ``KafkaSharedConfiguration/SecurityProtocol`` used to communicate with brokers.
     public var securityProtocol: KafkaSharedConfiguration.SecurityProtocol {
         get { self.getSecurityProtocol() ?? .plaintext }
         set { self.dictionary["security.protocol"] = newValue.description }
@@ -302,7 +335,8 @@ public struct KafkaConsumerConfig: Hashable, Equatable {
     }
 
     public init(
-        groupID: String = "",
+        consumptionStrategy: KafkaSharedConfiguration.ConsumptionStrategy,
+        backPressureStrategy: KafkaSharedConfiguration.BackPressureStrategy = .watermark(low: 10, high: 50),
         sessionTimeoutMs: UInt = 45000,
         heartbeatIntervalMs: UInt = 3000,
         maxPollInvervalMs: UInt = 300_000,
@@ -347,7 +381,10 @@ public struct KafkaConsumerConfig: Hashable, Equatable {
         saslUsername: String? = nil,
         saslPassword: String? = nil
     ) {
-        self.groupID = groupID
+        self._consumptionStrategy = consumptionStrategy
+        self.consumptionStrategy = consumptionStrategy // used to invoke set { } method
+        self.backPressureStrategy = backPressureStrategy
+
         self.sessionTimeoutMs = sessionTimeoutMs
         self.heartbeatIntervalMs = heartbeatIntervalMs
         self.maxPollInvervalMs = maxPollInvervalMs
@@ -433,9 +470,70 @@ public struct KafkaConsumerConfig: Hashable, Equatable {
     }
 }
 
-// MARK: - ConfigEnums + AutoOffsetReset
+// MARK: - KafkaSharedConfiguration + Consumer Additions
 
 extension KafkaSharedConfiguration {
+    /// A struct representing different back pressure strategies for consuming messages in ``KafkaConsumer``.
+    public struct BackPressureStrategy: Hashable, Equatable {
+        enum _BackPressureStrategy: Hashable, Equatable {
+            case watermark(low: Int, high: Int)
+        }
+
+        let _internal: _BackPressureStrategy
+
+        private init(backPressureStrategy: _BackPressureStrategy) {
+            self._internal = backPressureStrategy
+        }
+
+        /// A back pressure strategy based on high and low watermarks.
+        ///
+        /// The consumer maintains a buffer size between a low watermark and a high watermark
+        /// to control the flow of incoming messages.
+        ///
+        /// - Parameter low: The lower threshold for the buffer size (low watermark).
+        /// - Parameter high: The upper threshold for the buffer size (high watermark).
+        public static func watermark(low: Int, high: Int) -> BackPressureStrategy {
+            return .init(backPressureStrategy: .watermark(low: low, high: high))
+        }
+    }
+
+    /// A struct representing the different Kafka message consumption strategies.
+    public struct ConsumptionStrategy: Hashable, Equatable {
+        enum _ConsumptionStrategy: Hashable, Equatable {
+            case partition(topic: String, partition: KafkaPartition, offset: Int)
+            case group(groupID: String, topics: [String])
+        }
+
+        let _internal: _ConsumptionStrategy
+
+        private init(consumptionStrategy: _ConsumptionStrategy) {
+            self._internal = consumptionStrategy
+        }
+
+        /// A consumption strategy based on partition assignment.
+        /// The consumer reads from a specific partition of a topic at a given offset.
+        ///
+        /// - Parameter topic: The name of the Kafka topic.
+        /// - Parameter partition: The partition of the topic to consume from.
+        /// - Parameter offset: The offset to start consuming from.
+        public static func partition(
+            topic: String,
+            partition: KafkaPartition,
+            offset: Int = Int(RD_KAFKA_OFFSET_END)
+        ) -> ConsumptionStrategy {
+            return .init(consumptionStrategy: .partition(topic: topic, partition: partition, offset: offset))
+        }
+
+        /// A consumption strategy based on consumer group membership.
+        /// The consumer joins a consumer group identified by a group ID and consumes from multiple topics.
+        ///
+        /// - Parameter groupID: The ID of the consumer group to join.
+        /// - Parameter topics: An array of topic names to consume from.
+        public static func group(groupID: String, topics: [String]) -> ConsumptionStrategy {
+            return .init(consumptionStrategy: .group(groupID: groupID, topics: topics))
+        }
+    }
+
     /// 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/Configuration/KafkaProducerConfig.swift

@@ -199,7 +199,7 @@ public struct KafkaProducerConfig: Hashable, Equatable {
         set { self.dictionary["broker.address.ttl"] = String(newValue) }
     }
 
-    /// Allowed broker ``ConfigEnums/IPAddressFamily``.
+    /// Allowed broker ``KafkaSharedConfiguration/IPAddressFamily``.
     public var brokerAddressFamily: KafkaSharedConfiguration.IPAddressFamily {
         get { self.getIPAddressFamily() ?? .any }
         set { self.dictionary["broker.address.family"] = newValue.description }
@@ -217,7 +217,7 @@ public struct KafkaProducerConfig: Hashable, Equatable {
         set { self.dictionary["reconnect.backoff.max.ms"] = String(newValue) }
     }
 
-    /// ``ConfigEnums/SecurityProtocol`` used to communicate with brokers.
+    /// ``KafkaSharedConfiguration/SecurityProtocol`` used to communicate with brokers.
     public var securityProtocol: KafkaSharedConfiguration.SecurityProtocol {
         get { self.getSecurityProtocol() ?? .plaintext }
         set { self.dictionary["security.protocol"] = newValue.description }

Sources/SwiftKafka/Configuration/KafkaTopicConfig.swift

@@ -34,7 +34,7 @@ public struct KafkaTopicConfig: Hashable, Equatable {
         set { self.dictionary["message.timeout.ms"] = String(newValue) }
     }
 
-    /// Paritioner. See ``ConfigEnums/Partitioner`` for more information.
+    /// Paritioner. See ``KafkaSharedConfiguration/Partitioner`` for more information.
     public var partitioner: KafkaSharedConfiguration.Partitioner {
         get { self.getPartitioner() ?? .consistentRandom }
         set { self.dictionary["partitioner"] = newValue.description }

Sources/SwiftKafka/KafkaAcknowledgedMessage.swift

@@ -30,7 +30,7 @@ public struct KafkaAcknowledgedMessage: Hashable {
     /// The body of the message.
     public var value: ByteBuffer
     /// The offset of the message in its partition.
-    public var offset: Int64
+    public var offset: Int
 
     /// Initialize ``KafkaAcknowledgedMessage`` from `rd_kafka_message_t` pointer.
     /// - Throws: A ``KafkaAcknowledgedMessageError`` for failed acknowledgements or malformed messages.
@@ -63,6 +63,6 @@ public struct KafkaAcknowledgedMessage: Hashable {
             self.key = nil
         }
 
-        self.offset = Int64(rdKafkaMessage.offset)
+        self.offset = Int(rdKafkaMessage.offset)
     }
 }