KafkaClient.consumerClose: make non-blocking

Motivation:

[rd_kakfa_consumer_close](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a37b54d329e12d745889defe96e7d043d)
was blocking. This PR proposes using the
[rd_kakfa_consumer_close_queue](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a9dd5c18bdfed81c8847b259f0a8d498d)
API which is non-blocking and served through the normal poll loop.
We now
Modifications:

* `KafkaClient.consumerClose`: use `rd_kakfa_consumer_close_queue` in
  favour of `rd_kakfa_consumer_close`
* create a new variable `KafkaClient.isConsumerClosed` that indicates
  if the poll loop needs to continue polling or if it can stop running
* updated state management in `KafkaConsumer` to accomodate for polling
  when the `KafkaConsumer` is in the process of closing

Result:

Calling `KafkaClient.consumerClose` is not blocking anymore.

Author: felixschlegel

Sources/SwiftKafka/KafkaClient.swift

@@ -118,14 +118,6 @@ final class KafkaClient {
         }
     }
 
-    /// Close the consumer.
-    func consumerClose() throws {
-        let result = rd_kafka_consumer_close(self.kafkaHandle)
-        if result != RD_KAFKA_RESP_ERR_NO_ERROR {
-            throw KafkaError.rdKafkaError(wrapping: result)
-        }
-    }
-
     /// Wraps a Swift closure inside of a class to be able to pass it to `librdkafka` as an `OpaquePointer`.
     /// This is specifically used to pass a Swift closure as a commit callback for the ``KafkaConsumer``.
     final class CapturedCommitCallback {
@@ -205,6 +197,23 @@ final class KafkaClient {
         }
     }
 
+    /// 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`.
+    func consumerClose() throws {
+        let consumerQueue = rd_kafka_queue_get_consumer(self.kafkaHandle)
+        let result = rd_kafka_consumer_close_queue(self.kafkaHandle, consumerQueue)
+        let kafkaError = rd_kafka_error_code(result)
+        if kafkaError != RD_KAFKA_RESP_ERR_NO_ERROR {
+            throw KafkaError.rdKafkaError(wrapping: kafkaError)
+        }
+    }
+
+    var isConsumerClosed: Bool {
+        rd_kafka_consumer_closed(self.kafkaHandle) == 1
+    }
+
     /// Scoped accessor that enables safe access to the pointer of the client's Kafka handle.
     /// - Warning: Do not escape the pointer from the closure for later use.
     /// - Parameter body: The closure will use the Kafka handle pointer.

Sources/SwiftKafka/KafkaConsumer.swift

@@ -201,16 +201,20 @@ public final class KafkaConsumer {
             switch nextAction {
             case .pollForAndYieldMessage(let client, let source):
                 do {
-                    guard let message = try client.consumerPoll() else {
-                        break
+                    if let message = try client.consumerPoll() {
+                        // We do not support back pressure, we can ignore the yield result
+                        _ = source.yield(message)
                     }
-                    // We do not support back pressure, we can ignore the yield result
-                    _ = source.yield(message)
                 } catch {
                     source.finish()
                     throw error
                 }
                 try await Task.sleep(for: self.config.pollInterval)
+            case .pollUntilClosed(let client):
+                // Ignore poll result, we are closing down and just polling to commit
+                // outstanding consumer state
+                _ = try client.consumerPoll()
+                try await Task.sleep(for: self.config.pollInterval)
             case .terminatePollLoop:
                 return
             }
@@ -304,7 +308,9 @@ extension KafkaConsumer {
                 source: Producer.Source
             )
             /// The ``KafkaConsumer`` has been closed.
-            case finished
+            ///
+            /// - Parameter client: Client used for handling the connection to the Kafka cluster.
+            case finished(client: KafkaClient)
         }
 
         /// The current state of the StateMachine.
@@ -335,6 +341,11 @@ extension KafkaConsumer {
                 client: KafkaClient,
                 source: Producer.Source
             )
+            /// The ``KafkaConsumer`` is in the process of closing down, but still needs to poll
+            /// to commit its state to the broker.
+            ///
+            /// - Parameter client: Client used for handling the connection to the Kafka cluster.
+            case pollUntilClosed(client: KafkaClient)
             /// Terminate the poll loop.
             case terminatePollLoop
         }
@@ -351,8 +362,12 @@ extension KafkaConsumer {
                 fatalError("Subscribe to consumer group / assign to topic partition pair before reading messages")
             case .consuming(let client, let source):
                 return .pollForAndYieldMessage(client: client, source: source)
-            case .finished:
-                return .terminatePollLoop
+            case .finished(let client):
+                if client.isConsumerClosed {
+                    return .terminatePollLoop
+                } else {
+                    return .pollUntilClosed(client: client)
+                }
             }
         }
 
@@ -433,7 +448,7 @@ extension KafkaConsumer {
             case .initializing:
                 fatalError("subscribe() / assign() should have been invoked before \(#function)")
             case .consuming(let client, let source):
-                self.state = .finished
+                self.state = .finished(client: client)
                 return .shutdownGracefullyAndFinishSource(
                     client: client,
                     source: source