Allow groupID to be specified when assigning partition

Motivation:

A Consumer Group can provide a lot of benefits even if the
dynamic loadbalancing features are not used.

Modifications:

Allow for an optional GroupID when creating a partition
consumer.

Result:

Consumer Groups can now be used when manual assignment is
used.

Author: Ómar Kjartan Yasin

Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift

@@ -57,7 +57,7 @@ public struct KafkaConsumerConfiguration {
     /// A struct representing the different Kafka message consumption strategies.
     public struct ConsumptionStrategy: Sendable, Hashable {
         enum _ConsumptionStrategy: Sendable, Hashable {
-            case partition(topic: String, partition: KafkaPartition, offset: KafkaOffset)
+            case partition(groupID: String? = nil, topic: String, partition: KafkaPartition, offset: KafkaOffset)
             case group(groupID: String, topics: [String])
         }
 
@@ -82,6 +82,23 @@ public struct KafkaConsumerConfiguration {
             return .init(consumptionStrategy: .partition(topic: topic, partition: partition, offset: offset))
         }
 
+        /// A consumption strategy based on partition assignment.
+        /// The consumer reads from a specific partition of a topic at a given offset.
+        ///
+        /// - Parameters:
+        ///     - partition: The partition of the topic to consume from.
+        ///     - group: The ID of the consumer group to commit to.
+        ///     - topic: The name of the Kafka topic.
+        ///     - offset: The offset to start consuming from. Defaults to the end of the Kafka partition queue (meaning wait for the next produced message).
+        public static func partition(
+            _ partition: KafkaPartition,
+            groupID: String,
+            topic: String,
+            offset: KafkaOffset = .end
+        ) -> ConsumptionStrategy {
+            return .init(consumptionStrategy: .partition(groupID: groupID, 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.
         ///
@@ -261,12 +278,17 @@ extension KafkaConsumerConfiguration {
         var resultDict: [String: String] = [:]
 
         switch self.consumptionStrategy._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
-            resultDict["group.id"] = UUID().uuidString
+        case .partition(groupID: let groupID, topic: _, partition: _, offset: _):
+            if let groupID = groupID {
+                resultDict["group.id"] = groupID
+            } else {
+                // 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
+                resultDict["group.id"] = UUID().uuidString
+            }
+            
         case .group(groupID: let groupID, topics: _):
             resultDict["group.id"] = groupID
         }

Sources/Kafka/KafkaConsumer.swift

@@ -359,7 +359,7 @@ public final class KafkaConsumer: Sendable, Service {
 
     private func _run() async throws {
         switch self.configuration.consumptionStrategy._internal {
-        case .partition(topic: let topic, partition: let partition, offset: let offset):
+        case .partition(groupID: _, topic: let topic, partition: let partition, offset: let offset):
             try self.assign(topic: topic, partition: partition, offset: offset)
         case .group(groupID: _, topics: let topics):
             try self.subscribe(topics: topics)