KafkaProducer: replace callbacks with eventPoll (#82)

* `KafkaProducer`: replace callbacks with `eventPoll`

Motivation:

Currently we serve `librdkafka` logs and delivery reports through
callbacks. In the future we also want to get notified about `librdkafka`
triggered errors, statistics and rebalances. As having the callbacks
adds a lot of overhead e.g. by having `Unmanaged` references and having
to wrap Swift closures in C callbacks, we rather want to have an
`eventPoll` method that regularly polls for all sorts of events and
handles them inline.

Modifications:

* `KafkaClient`
    * remove `poll` method
    * add new method `eventPoll`
    * move mapping from `librdkafka` logs to `swift-log` to
      `KafkaClient.eventPoll`
* `KafkaProducer`:
    * receive acknowledgements through `KafkaClient.eventPoll` instead of using
      `KafkaClient.poll` to trigger event callbacks
    * replace `deliveryReport` callback and handle deliveryReport event directly in
      `KafkaProducer`
* `RDKafkaConfig`
    * remove `logging` logic
* `KafkaProducerTests`:
    * add `testProducerLog` that asserts that `librdkafka` logs get
      forwarded to `swift-log`
* move `convertMessageToAcknowledgementResult` from `RDKafkaConfig` to
  `KafkaClient`

* Fix: don't listen to RD_KAFKA_EVENT_DR in non-ack producer

* Review Franz

Modifications:

* `KafkaClient.eventPoll`:
    * reserve capacity for arrays to avoid
    reallocations
    * move `rd_kafka_event_destory` to `defer` statement
    * get rid of named loop (just return from loop)
    * wrap handling of different event types in `private` methods
* wrap subset of `RD_KAFKA_*` event types in Swift type `RDKafkaEvent`

Author: felixschlegel

Sources/SwiftKafka/KafkaClient.swift

@@ -23,19 +23,22 @@ final class KafkaClient: Sendable {
 
     /// Handle for the C library's Kafka instance.
     private let kafkaHandle: OpaquePointer
-    /// References the opaque object passed to the config to ensure ARC retains it as long as the client exists.
-    private let opaque: RDKafkaConfig.CapturedClosures
     /// A logger.
     private let logger: Logger
 
+    /// `librdkafka`'s main `rd_kafka_queue_t`.
+    private let mainQueue: OpaquePointer
+
     init(
         kafkaHandle: OpaquePointer,
-        opaque: RDKafkaConfig.CapturedClosures,
         logger: Logger
     ) {
         self.kafkaHandle = kafkaHandle
-        self.opaque = opaque
         self.logger = logger
+
+        self.mainQueue = rd_kafka_queue_get_main(self.kafkaHandle)
+
+        rd_kafka_set_log_queue(self.kafkaHandle, self.mainQueue)
     }
 
     deinit {
@@ -83,18 +86,101 @@ final class KafkaClient: Sendable {
         }
     }
 
-    /// Polls the Kafka client for events.
+    /// Swift wrapper for events from `librdkafka`'s event queue.
+    enum KafkaEvent {
+        case deliveryReport(results: [Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>])
+    }
+
+    /// Poll the event `rd_kafka_queue_t` for new events.
     ///
-    /// Events will cause application-provided callbacks to be called.
+    /// - Parameter maxEvents:Maximum number of events to serve in one invocation.
+    func eventPoll(maxEvents: Int = 100) -> [KafkaEvent] {
+        var events = [KafkaEvent]()
+        events.reserveCapacity(maxEvents)
+
+        for _ in 0..<maxEvents {
+            let event = rd_kafka_queue_poll(self.mainQueue, 0)
+            defer { rd_kafka_event_destroy(event) }
+
+            let rdEventType = rd_kafka_event_type(event)
+            guard let eventType = RDKafkaEvent(rawValue: rdEventType) else {
+                fatalError("Unsupported event type: \(rdEventType)")
+            }
+
+            switch eventType {
+            case .deliveryReport:
+                let forwardEvent = self.handleDeliveryReportEvent(event)
+                events.append(forwardEvent) // Return KafkaEvent.deliveryReport as part of this method
+            case .log:
+                self.handleLogEvent(event)
+            case .none:
+                // Finished reading events, return early
+                return events
+            default:
+                break // Ignored Event
+            }
+        }
+
+        return events
+    }
+
+    /// Handle event of type `RDKafkaEvent.deliveryReport`.
     ///
-    /// - Parameter timeout: Specifies the maximum amount of time
-    /// (in milliseconds) that the call will block waiting for events.
-    /// For non-blocking calls, provide 0 as `timeout`.
-    /// To wait indefinitely for an event, provide -1.
-    /// - Returns: The number of events served.
-    @discardableResult
-    func poll(timeout: Int32) -> Int32 {
-        return rd_kafka_poll(self.kafkaHandle, timeout)
+    /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.
+    /// - Returns: `KafkaEvent` to be returned as part of ``KafkaClient.eventPoll()`.
+    private func handleDeliveryReportEvent(_ event: OpaquePointer?) -> KafkaEvent {
+        let deliveryReportCount = rd_kafka_event_message_count(event)
+        var deliveryReportResults = [Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>]()
+        deliveryReportResults.reserveCapacity(deliveryReportCount)
+
+        while let messagePointer = rd_kafka_event_message_next(event) {
+            guard let message = Self.convertMessageToAcknowledgementResult(messagePointer: messagePointer) else {
+                continue
+            }
+            deliveryReportResults.append(message)
+        }
+
+        return .deliveryReport(results: deliveryReportResults)
+    }
+
+    /// Handle event of type `RDKafkaEvent.log`.
+    ///
+    /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.
+    private func handleLogEvent(_ event: OpaquePointer?) {
+        var faculty: UnsafePointer<CChar>?
+        var buffer: UnsafePointer<CChar>?
+        var level: Int32 = 0
+        if rd_kafka_event_log(event, &faculty, &buffer, &level) == 0 {
+            if let faculty, let buffer {
+                // Mapping according to https://en.wikipedia.org/wiki/Syslog
+                switch level {
+                case 0...2: /* Emergency, Alert, Critical */
+                    self.logger.critical(
+                        Logger.Message(stringLiteral: String(cString: buffer)), source: String(cString: faculty)
+                    )
+                case 3: /* Error */
+                    self.logger.error(
+                        Logger.Message(stringLiteral: String(cString: buffer)), source: String(cString: faculty)
+                    )
+                case 4: /* Warning */
+                    self.logger.warning(
+                        Logger.Message(stringLiteral: String(cString: buffer)), source: String(cString: faculty)
+                    )
+                case 5: /* Notice */
+                    self.logger.notice(
+                        Logger.Message(stringLiteral: String(cString: buffer)), source: String(cString: faculty)
+                    )
+                case 6: /* Informational */
+                    self.logger.info(
+                        Logger.Message(stringLiteral: String(cString: buffer)), source: String(cString: faculty)
+                    )
+                default: /* Debug */
+                    self.logger.debug(
+                        Logger.Message(stringLiteral: String(cString: buffer)), source: String(cString: faculty)
+                    )
+                }
+            }
+        }
     }
 
     /// Redirect the main ``KafkaClient/poll(timeout:)`` queue to the `KafkaConsumer`'s
@@ -271,4 +357,30 @@ final class KafkaClient: Sendable {
     func withKafkaHandlePointer<T>(_ body: (OpaquePointer) throws -> T) rethrows -> T {
         return try body(self.kafkaHandle)
     }
+
+    /// Convert an unsafe`rd_kafka_message_t` object to a safe ``KafkaAcknowledgementResult``.
+    /// - Parameter messagePointer: An `UnsafePointer` pointing to the `rd_kafka_message_t` object in memory.
+    /// - Returns: A ``KafkaAcknowledgementResult``.
+    private static func convertMessageToAcknowledgementResult(
+        messagePointer: UnsafePointer<rd_kafka_message_t>?
+    ) -> Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>? {
+        guard let messagePointer else {
+            return nil
+        }
+
+        let messageID = KafkaProducerMessageID(rawValue: UInt(bitPattern: messagePointer.pointee._private))
+
+        let messageResult: Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>
+        do {
+            let message = try KafkaAcknowledgedMessage(messagePointer: messagePointer, id: messageID)
+            messageResult = .success(message)
+        } catch {
+            guard let error = error as? KafkaAcknowledgedMessageError else {
+                fatalError("Caught error that is not of type \(KafkaAcknowledgedMessageError.self)")
+            }
+            messageResult = .failure(error)
+        }
+
+        return messageResult
+    }
 }

Sources/SwiftKafka/KafkaConsumer.swift

@@ -73,7 +73,12 @@ public final class KafkaConsumer: Sendable, Service {
         self.config = config
         self.logger = logger
 
-        let client = try RDKafka.createClient(type: .consumer, configDictionary: config.dictionary, logger: logger)
+        let client = try RDKafka.createClient(
+            type: .consumer,
+            configDictionary: config.dictionary,
+            events: [.log],
+            logger: logger
+        )
 
         self.stateMachine = NIOLockedValueBox(StateMachine(logger: self.logger))
 

Sources/SwiftKafka/KafkaProducer.swift

@@ -98,9 +98,7 @@ public final class KafkaProducer: Service, Sendable {
         let client = try RDKafka.createClient(
             type: .producer,
             configDictionary: config.dictionary,
-            // Having no callback will discard any incoming acknowledgement messages
-            // Ref: rdkafka_broker.c:rd_kafka_dr_msgq
-            deliveryReportCallback: nil,
+            events: [.log], // No .deliveryReport here!
             logger: logger
         )
 
@@ -149,15 +147,7 @@ public final class KafkaProducer: Service, Sendable {
         let client = try RDKafka.createClient(
             type: .producer,
             configDictionary: config.dictionary,
-            deliveryReportCallback: { [logger, source] messageResult in
-                guard let messageResult else {
-                    logger.error("Could not resolve acknowledged message")
-                    return
-                }
-
-                // Ignore YieldResult as we don't support back pressure in KafkaProducer
-                _ = source.yield(messageResult)
-            },
+            events: [.log, .deliveryReport],
             logger: logger
         )
 
@@ -193,8 +183,15 @@ public final class KafkaProducer: Service, Sendable {
         while !Task.isCancelled {
             let nextAction = self.stateMachine.withLockedValue { $0.nextPollLoopAction() }
             switch nextAction {
-            case .poll(let client):
-                client.poll(timeout: 0)
+            case .poll(let client, let source):
+                let events = client.eventPoll()
+                for event in events {
+                    switch event {
+                    case .deliveryReport(let results):
+                        // Ignore YieldResult as we don't support back pressure in KafkaProducer
+                        results.forEach { _ = source?.yield($0) }
+                    }
+                }
                 try await Task.sleep(for: self.config.pollInterval)
             case .terminatePollLoopAndFinishSource(let source):
                 source?.finish()
@@ -297,7 +294,10 @@ extension KafkaProducer {
         /// Action to be taken when wanting to poll.
         enum PollLoopAction {
             /// Poll client for new consumer messages.
-            case poll(client: KafkaClient)
+            ///
+            /// - Parameter client: Client used for handling the connection to the Kafka cluster.
+            /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
+            case poll(client: KafkaClient, source: Producer.Source?)
             /// Terminate the poll loop and finish the given `NIOAsyncSequenceProducerSource`.
             ///
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
@@ -314,11 +314,11 @@ extension KafkaProducer {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
-            case .started(let client, _, _, _):
-                return .poll(client: client)
+            case .started(let client, _, let source, _):
+                return .poll(client: client, source: source)
             case .flushing(let client, let source):
                 if client.outgoingQueueSize > 0 {
-                    return .poll(client: client)
+                    return .poll(client: client, source: source)
                 } else {
                     self.state = .finished
                     return .terminatePollLoopAndFinishSource(source: source)

Sources/SwiftKafka/RDKafka/RDKafka.swift

@@ -27,23 +27,14 @@ struct RDKafka {
     static func createClient(
         type: ClientType,
         configDictionary: [String: String],
-        deliveryReportCallback: RDKafkaConfig.CapturedClosures.DeliveryReportClosure? = nil,
+        events: [RDKafkaEvent],
         logger: Logger
     ) throws -> KafkaClient {
         let clientType = type == .producer ? RD_KAFKA_PRODUCER : RD_KAFKA_CONSUMER
 
         let rdConfig = try RDKafkaConfig.createFrom(configDictionary: configDictionary)
-
-        // Check that delivery report callback can be only set for producer
-        guard deliveryReportCallback == nil || type == .producer else {
-            fatalError("Delivery report callback can't be defined for consumer client")
-        }
-
-        let opaque = RDKafkaConfig.setCallbackClosures(
-            configPointer: rdConfig,
-            deliveryReportCallback: deliveryReportCallback,
-            logger: logger
-        )
+        try RDKafkaConfig.set(configPointer: rdConfig, key: "log.queue", value: "true")
+        RDKafkaConfig.setEvents(configPointer: rdConfig, events: events)
 
         let errorChars = UnsafeMutablePointer<CChar>.allocate(capacity: KafkaClient.stringSize)
         defer { errorChars.deallocate() }
@@ -61,6 +52,6 @@ struct RDKafka {
             throw KafkaError.client(reason: errorString)
         }
 
-        return KafkaClient(kafkaHandle: handle, opaque: opaque, logger: logger)
+        return KafkaClient(kafkaHandle: handle, logger: logger)
     }
 }