Fix: Kafka consumer leaves group after 5 mins (#113)

* Fix: Kafka consumer leaves group after 5 mins

Motivation:

This PR fixes issue #110.

`KakfaConsumer`: by polling the `rd_kafka_queue_get_main` queue instead of the `rd_kafka_queue_get_consumer` queue,
the timer for `max.poll.interval.ms` did not get reset which eventually
resulted in a timeout despite polling. (See [`librdkafka`
documentation](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#acacdb55ae7cb6abfbde89621e512b078))

Modifications:

* `RDKafkaClient`:
    * rename `mainQueue` to `queue`
    * use `rd_kafka_queue_get_consumer` instead of
      `rd_kafka_queue_get_main` for `KakfaConsumer` clients
      -> this will reset the timer for `max.poll.interval.ms` so that
         the consumer does not time out despite polling
    * invoke `rd_kafka_queue_destroy(self.queue)` on
      `RDKafkaClient.deinit` to loose reference to queue

* Review blindspot

Modifications:

* update comment at invocation of `RDKafkaClient.pollSetConsumer`
* don't fail softly when `rd_kafka_queue_get_consumer` returns `nil`
* don't create new reference to consumer queue in
  `RDKafkaClient.consumerClose()`

Author: felixschlegel

Sources/Kafka/KafkaConsumer.swift

@@ -170,8 +170,7 @@ public final class KafkaConsumer: Sendable, Service {
             )
         }
 
-        // Events that would be triggered by ``RDKafkaClient/poll(timeout:)``
-        // are now triggered by ``RDKafkaClient/consumerPoll``.
+        // Forward main queue events to the consumer queue.
         try client.pollSetConsumer()
 
         switch configuration.consumptionStrategy._internal {

Sources/Kafka/RDKafka/RDKafkaClient.swift

@@ -19,37 +19,54 @@ import Logging
 /// Base class for ``KafkaProducer`` and ``KafkaConsumer``,
 /// which is used to handle the connection to the Kafka ecosystem.
 final class RDKafkaClient: Sendable {
+    // Default size for Strings returned from C API
+    static let stringSize = 1024
+
     /// Determines if client is a producer or a consumer.
     enum ClientType {
         case producer
         case consumer
     }
 
-    // Default size for Strings returned from C API
-    static let stringSize = 1024
-
     /// Handle for the C library's Kafka instance.
     private let kafkaHandle: OpaquePointer
     /// A logger.
     private let logger: Logger
 
-    /// `librdkafka`'s main `rd_kafka_queue_t`.
-    private let mainQueue: OpaquePointer
+    /// `librdkafka`'s `rd_kafka_queue_t` that events are received on.
+    private let queue: OpaquePointer
 
     // Use factory method to initialize
     private init(
+        type: ClientType,
         kafkaHandle: OpaquePointer,
         logger: Logger
     ) {
         self.kafkaHandle = kafkaHandle
         self.logger = logger
 
-        self.mainQueue = rd_kafka_queue_get_main(self.kafkaHandle)
+        if type == .consumer {
+            if let consumerQueue = rd_kafka_queue_get_consumer(self.kafkaHandle) {
+                // (Important)
+                // Polling the queue counts as a consumer poll, and will reset the timer for `max.poll.interval.ms`.
+                self.queue = consumerQueue
+            } else {
+                fatalError("""
+                Internal error: failed to get consumer queue. \
+                A group.id should be set even when the client is not part of a consumer group. \
+                See https://github.com/edenhill/librdkafka/issues/3261 for more information.
+                """)
+            }
+        } else {
+            self.queue = rd_kafka_queue_get_main(self.kafkaHandle)
+        }
 
-        rd_kafka_set_log_queue(self.kafkaHandle, self.mainQueue)
+        rd_kafka_set_log_queue(self.kafkaHandle, self.queue)
     }
 
     deinit {
+        // Loose reference to librdkafka's event queue
+        rd_kafka_queue_destroy(self.queue)
         rd_kafka_destroy(kafkaHandle)
     }
 
@@ -60,8 +77,6 @@ final class RDKafkaClient: Sendable {
         events: [RDKafkaEvent],
         logger: Logger
     ) throws -> RDKafkaClient {
-        let clientType = type == .producer ? RD_KAFKA_PRODUCER : RD_KAFKA_CONSUMER
-
         let rdConfig = try RDKafkaConfig.createFrom(configDictionary: configDictionary)
         // Manually override some of the configuration options
         // Handle logs in event queue
@@ -75,6 +90,7 @@ final class RDKafkaClient: Sendable {
         let errorChars = UnsafeMutablePointer<CChar>.allocate(capacity: RDKafkaClient.stringSize)
         defer { errorChars.deallocate() }
 
+        let clientType = type == .producer ? RD_KAFKA_PRODUCER : RD_KAFKA_CONSUMER
         guard let handle = rd_kafka_new(
             clientType,
             rdConfig,
@@ -88,7 +104,7 @@ final class RDKafkaClient: Sendable {
             throw KafkaError.client(reason: errorString)
         }
 
-        return RDKafkaClient(kafkaHandle: handle, logger: logger)
+        return RDKafkaClient(type: type, kafkaHandle: handle, logger: logger)
     }
 
     /// Produce a message to the Kafka cluster.
@@ -163,7 +179,7 @@ final class RDKafkaClient: Sendable {
         events.reserveCapacity(maxEvents)
 
         for _ in 0..<maxEvents {
-            let event = rd_kafka_queue_poll(self.mainQueue, 0)
+            let event = rd_kafka_queue_poll(self.queue, 0)
             defer { rd_kafka_event_destroy(event) }
 
             let rdEventType = rd_kafka_event_type(event)
@@ -399,7 +415,7 @@ final class RDKafkaClient: Sendable {
                 rd_kafka_commit_queue(
                     self.kafkaHandle,
                     listPointer,
-                    self.mainQueue,
+                    self.queue,
                     nil,
                     opaquePointer
                 )
@@ -433,8 +449,7 @@ final class RDKafkaClient: Sendable {
     ///
     /// Make sure to run poll loop until ``RDKafkaClient/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 result = rd_kafka_consumer_close_queue(self.kafkaHandle, self.queue)
         let kafkaError = rd_kafka_error_code(result)
         if kafkaError != RD_KAFKA_RESP_ERR_NO_ERROR {
             throw KafkaError.rdKafkaError(wrapping: kafkaError)