Add dynamic subscribe, unsubscribe APIs to KafkaConsumer (#227)

## Motivation

Topics are locked at initialization time via `consumptionStrategy` with
no way to change subscriptions at runtime. The existing private
`subscribe()` crashes on re-subscribe (`fatalError`), and
`consumptionStrategy` is force-unwrapped in `_run()` which crashes if
nil.

Applications that need dynamic topic routing, multi-tenant consumption,
or config-driven subscriptions currently cannot use this client without
recreating the consumer. This aligns the Swift client with the standard
Kafka client contract where subscribe/unsubscribe are runtime
operations, not initialization-time configuration.

## Summary

- Add public `subscribe(topics:)`, `unsubscribe()`, and `subscription()`
on `KafkaConsumer`
- `subscribe(topics:)` replaces the current subscription, matching
librdkafka semantics
- Add `unsubscribe()` and `subscription()` to `RDKafkaClient`
- Fix force-unwrap crash when `consumptionStrategy` is nil
- Consumers can be created without `consumptionStrategy` and subscribe
dynamically

## Test plan
- [x] Unit tests: closed consumer, empty topics, subscribe before run,
nil strategy, duplicate topics, regex patterns
- [x] Integration tests: subscription replacement, unsubscribe,
re-subscribe, dynamic multi-topic consume
- [x] `swift test --skip IntegrationTests` passes
- [x] `make docker-test` passes

Author: ssikka100

Sources/Kafka/KafkaConsumer.swift

@@ -334,11 +334,73 @@ public final class KafkaConsumer: Sendable, Service {
         )
     }
 
-    /// Subscribe to the given list of `topics`.
-    /// The partition assignment happens automatically using `KafkaConsumer`'s consumer group.
+    // MARK: - Subscription Management
+
+    /// Subscribe to the given list of topics.
+    ///
+    /// This replaces any previous subscription. The partition assignment happens
+    /// automatically using the consumer's consumer group. Topic names prefixed
+    /// with `^` are treated as regular expressions. Passing an empty array is
+    /// equivalent to calling ``unsubscribe()``.
+    ///
     /// - 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 {
+    /// - Throws: A ``KafkaError`` if the consumer is closed or subscribing failed.
+    public func subscribe(topics: [String]) throws {
+        guard !topics.isEmpty else {
+            try self.unsubscribe()
+            return
+        }
+
+        let action = self.stateMachine.withLockedValue { $0.withClientForSubscription() }
+        switch action {
+        case .client(let client):
+            let subscription = RDKafkaTopicPartitionList()
+            for topic in topics {
+                subscription.add(
+                    topic: topic,
+                    partition: KafkaPartition.unassigned
+                )
+            }
+            try client.subscribe(topicPartitionList: subscription)
+        case .throwClosedError:
+            throw KafkaError.connectionClosed(reason: "Consumer is closed")
+        }
+    }
+
+    /// Unsubscribe from the current subscription.
+    ///
+    /// Clears all topic subscriptions. The consumer leaves the consumer group
+    /// and stops receiving messages. This triggers a rebalance event.
+    /// Can re-subscribe later with ``subscribe(topics:)``.
+    ///
+    /// - Throws: A ``KafkaError`` if the consumer is closed or unsubscribing failed.
+    public func unsubscribe() throws {
+        let action = self.stateMachine.withLockedValue { $0.withClientForSubscription() }
+        switch action {
+        case .client(let client):
+            try client.unsubscribe()
+        case .throwClosedError:
+            throw KafkaError.connectionClosed(reason: "Consumer is closed")
+        }
+    }
+
+    /// Get the current topic subscription.
+    ///
+    /// - Returns: An array of topic names or patterns the consumer is currently subscribed to.
+    /// - Throws: A ``KafkaError`` if the consumer is closed or the query failed.
+    public func subscribedTopics() throws -> [String] {
+        let action = self.stateMachine.withLockedValue { $0.withClientForSubscription() }
+        switch action {
+        case .client(let client):
+            return try client.subscription()
+        case .throwClosedError:
+            throw KafkaError.connectionClosed(reason: "Consumer is closed")
+        }
+    }
+
+    /// Internal startup subscription that transitions the state machine from `.initializing` to `.running`.
+    /// Called once during `_run()` to set up the initial subscription from `consumptionStrategy`.
+    private func initialSubscribe(topics: [String]) throws {
         let action = self.stateMachine.withLockedValue { $0.setUpConnection() }
         switch action {
         case .setUpConnection(let client):
@@ -389,11 +451,23 @@ public final class KafkaConsumer: Sendable, Service {
     }
 
     private func _run() async throws {
-        switch self.config.consumptionStrategy!._internal {
-        case .partition(groupID: _, let topic, let partition, let offset):
-            try self.assign(topic: topic, partition: partition, offset: offset)
-        case .group(groupID: _, let topics):
-            try self.subscribe(topics: topics)
+        if let strategy = self.config.consumptionStrategy?._internal {
+            switch strategy {
+            case .partition(groupID: _, let topic, let partition, let offset):
+                try self.assign(topic: topic, partition: partition, offset: offset)
+            case .group(groupID: _, let topics):
+                try self.initialSubscribe(topics: topics)
+            }
+        } else {
+            // No consumptionStrategy set — user will call subscribe(topics:) manually.
+            // Transition state machine to .running so the event loop can start.
+            let action = self.stateMachine.withLockedValue { $0.setUpConnection() }
+            switch action {
+            case .setUpConnection:
+                break
+            case .consumerClosed:
+                throw KafkaError.connectionClosed(reason: "Consumer closed before run")
+            }
         }
         try await self.eventRunLoop()
     }
@@ -907,19 +981,32 @@ extension KafkaConsumer {
 
         /// Get the running client for performing an operation that requires an active consumer.
         ///
-        /// This is a general-purpose accessor used by methods that need the underlying
-        /// `RDKafkaClient` while the consumer is running (e.g., commit, storeOffset,
-        /// committed, position, seek, isAssignmentLost).
-        ///
         /// - Returns: The action to be taken.
-        ///
-        /// - Important: This function calls `fatalError` if called while in the `.initializing` state.
         func withClient() -> ClientAction {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .initializing:
-                fatalError("Subscribe to consumer group / assign to topic partition pair before reading messages")
+                return .throwClosedError
+            case .running(let client, _):
+                return .client(client)
+            case .finishing, .finished:
+                return .throwClosedError
+            }
+        }
+
+        /// Get the client for subscription management operations.
+        ///
+        /// Unlike ``withClient()``, this accessor works in both `.initializing` and `.running`
+        /// states, allowing subscribe/unsubscribe to be called at any time after consumer creation.
+        ///
+        /// - Returns: The action to be taken.
+        func withClientForSubscription() -> ClientAction {
+            switch self.state {
+            case .uninitialized:
+                fatalError("\(#function) invoked while still in state \(self.state)")
+            case .initializing(let client, _):
+                return .client(client)
             case .running(let client, _):
                 return .client(client)
             case .finishing, .finished:
@@ -935,16 +1022,18 @@ extension KafkaConsumer {
             case triggerGracefulShutdown(client: RDKafkaClient)
         }
 
-        /// Get action to be taken when wanting to do close the consumer.
-        /// - Returns: The action to be taken,  or `nil` if there is no action to be taken.
-        ///
-        /// - Important: This function throws a `fatalError` if called while in the `.initializing` state.
+        /// Get action to be taken when wanting to close the consumer.
+        /// - Returns: The action to be taken, or `nil` if there is no action to be taken.
         mutating func finish() -> FinishAction? {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .initializing:
-                fatalError("Subscribe to consumer group / assign to topic partition pair before reading messages")
+                // No poll loop is active in .initializing state. Since consumerClose()
+                // requires an active poll loop to process broker responses, we skip it
+                // and transition straight to .finished.
+                self.state = .finished
+                return nil
             case .running(let client, let rebalanceContext):
                 self.state = .finishing(client: client, rebalanceContext: rebalanceContext)
                 return .triggerGracefulShutdown(client: client)

Sources/Kafka/RDKafka/RDKafkaClient.swift

@@ -598,6 +598,36 @@ public final class RDKafkaClient: Sendable {
         }
     }
 
+    /// Unsubscribe from the current subscription set.
+    func unsubscribe() throws {
+        let result = rd_kafka_unsubscribe(self.kafkaHandle.pointer)
+        if result != RD_KAFKA_RESP_ERR_NO_ERROR {
+            throw KafkaError.rdKafkaError(wrapping: result)
+        }
+    }
+
+    /// Get the current topic subscription.
+    /// - Returns: An array of topic names the consumer is subscribed to.
+    func subscription() throws -> [String] {
+        var tpl: UnsafeMutablePointer<rd_kafka_topic_partition_list_t>?
+        let result = rd_kafka_subscription(self.kafkaHandle.pointer, &tpl)
+        if result != RD_KAFKA_RESP_ERR_NO_ERROR {
+            throw KafkaError.rdKafkaError(wrapping: result)
+        }
+
+        guard let list = tpl else { return [] }
+        defer { rd_kafka_topic_partition_list_destroy(list) }
+
+        var topics: [String] = []
+        for i in 0..<Int(list.pointee.cnt) {
+            let element = list.pointee.elems[i]
+            if let topicCString = element.topic {
+                topics.append(String(cString: topicCString))
+            }
+        }
+        return topics
+    }
+
     /// Atomic assignment of partitions to consume.
     /// - Parameter topicPartitionList: Pointer to a list of topics + partition pairs.
     func assign(topicPartitionList: RDKafkaTopicPartitionList) throws {