KafkaConsumer: replace callbacks with eventPoll (#86)

Motivation:

By using the polling for events we can get rid of most of our callbacks
and have the ability to conveniently listen to more `librdkafka` events like errors, statistics, etc. in the future.

Modifications:

* handle `RD_KAFKA_EVENT_FETCH` in `KafkaClient.eventPoll`
* handle `RD_KAFKA_EVENT_OFFSET_COMMIT` in `KafkaClient.eventPoll` -> we
  cannot have an event handler and callbacks served at the same time,
  that is why the event handler also invokes the consumer commit callback
* delete `KafkaClient.consumerPoll()`
* use `KafkaClient.eventPoll` instead of `KafkaClient.consumerPoll` to
  receive `KafkaConsumerMessage`s
* create `KafkaConsumerTests`
    * add `KafkaConsumerTests.testConsumerLog`
    * move `MockLogHandler` to `Utilities.swift`

Author: felixschlegel

Sources/SwiftKafka/KafkaClient.swift

@@ -89,6 +89,7 @@ final class KafkaClient: Sendable {
     /// Swift wrapper for events from `librdkafka`'s event queue.
     enum KafkaEvent {
         case deliveryReport(results: [Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>])
+        case consumerMessages(result: Result<KafkaConsumerMessage, Error>)
     }
 
     /// Poll the event `rd_kafka_queue_t` for new events.
@@ -110,9 +111,15 @@ final class KafkaClient: Sendable {
             switch eventType {
             case .deliveryReport:
                 let forwardEvent = self.handleDeliveryReportEvent(event)
-                events.append(forwardEvent) // Return KafkaEvent.deliveryReport as part of this method
+                events.append(forwardEvent)
+            case .fetch:
+                if let forwardEvent = self.handleFetchEvent(event) {
+                    events.append(forwardEvent)
+                }
             case .log:
                 self.handleLogEvent(event)
+            case .offsetCommit:
+                self.handleOffsetCommitEvent(event)
             case .none:
                 // Finished reading events, return early
                 return events
@@ -140,9 +147,30 @@ final class KafkaClient: Sendable {
             deliveryReportResults.append(message)
         }
 
+        // The returned message(s) MUST NOT be freed with rd_kafka_message_destroy().
         return .deliveryReport(results: deliveryReportResults)
     }
 
+    /// Handle event of type `RDKafkaEvent.fetch`.
+    ///
+    /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.
+    /// - Returns: `KafkaEvent` to be returned as part of ``KafkaClient.eventPoll()`.
+    private func handleFetchEvent(_ event: OpaquePointer?) -> KafkaEvent? {
+        do {
+            // RD_KAFKA_EVENT_FETCH only returns a single message:
+            // https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a3a855eb7bdf17f5797d4911362a5fc7c
+            if let messagePointer = rd_kafka_event_message_next(event) {
+                let message = try KafkaConsumerMessage(messagePointer: messagePointer)
+                return .consumerMessages(result: .success(message))
+            } else {
+                return nil
+            }
+        } catch {
+            return .consumerMessages(result: .failure(error))
+        }
+        // The returned message(s) MUST NOT be freed with rd_kafka_message_destroy().
+    }
+
     /// Handle event of type `RDKafkaEvent.log`.
     ///
     /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.
@@ -183,6 +211,25 @@ final class KafkaClient: Sendable {
         }
     }
 
+    /// Handle event of type `RDKafkaEvent.offsetCommit`.
+    ///
+    /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.
+    private func handleOffsetCommitEvent(_ event: OpaquePointer?) {
+        guard let opaquePointer = rd_kafka_event_opaque(event) else {
+            fatalError("Could not resolve reference to catpured Swift callback instance")
+        }
+        let opaque = Unmanaged<CapturedCommitCallback>.fromOpaque(opaquePointer).takeUnretainedValue()
+        let actualCallback = opaque.closure
+
+        let error = rd_kafka_event_error(event)
+        guard error == RD_KAFKA_RESP_ERR_NO_ERROR else {
+            let kafkaError = KafkaError.rdKafkaError(wrapping: error)
+            actualCallback(.failure(kafkaError))
+            return
+        }
+        actualCallback(.success(()))
+    }
+
     /// Redirect the main ``KafkaClient/poll(timeout:)`` queue to the `KafkaConsumer`'s
     /// queue (``KafkaClient/consumerPoll``).
     ///
@@ -197,32 +244,6 @@ final class KafkaClient: Sendable {
         }
     }
 
-    /// Request a new message from the Kafka cluster.
-    ///
-    /// - Important: This method should only be invoked from ``KafkaConsumer``.
-    ///
-    /// - Returns: A ``KafkaConsumerMessage`` or `nil` if there are no new messages.
-    /// - Throws: A ``KafkaError`` if the received message is an error message or malformed.
-    func consumerPoll() throws -> KafkaConsumerMessage? {
-        guard let messagePointer = rd_kafka_consumer_poll(self.kafkaHandle, 0) else {
-            // No error, there might be no more messages
-            return nil
-        }
-
-        defer {
-            // Destroy message otherwise poll() will block forever
-            rd_kafka_message_destroy(messagePointer)
-        }
-
-        // Reached the end of the topic+partition queue on the broker
-        if messagePointer.pointee.err == RD_KAFKA_RESP_ERR__PARTITION_EOF {
-            return nil
-        }
-
-        let message = try KafkaConsumerMessage(messagePointer: messagePointer)
-        return message
-    }
-
     /// Subscribe to topic set using balanced consumer groups.
     /// - Parameter topicPartitionList: Pointer to a list of topics + partition pairs.
     func subscribe(topicPartitionList: RDKafkaTopicPartitionList) throws {
@@ -285,39 +306,12 @@ final class KafkaClient: Sendable {
             // should not be counted in ARC as this can lead to memory leaks.
             let opaquePointer: UnsafeMutableRawPointer? = Unmanaged.passUnretained(capturedClosure).toOpaque()
 
-            let consumerQueue = rd_kafka_queue_get_consumer(self.kafkaHandle)
-
-            // Create a C closure that calls the captured closure
-            let callbackWrapper: (
-                @convention(c) (
-                    OpaquePointer?,
-                    rd_kafka_resp_err_t,
-                    UnsafeMutablePointer<rd_kafka_topic_partition_list_t>?,
-                    UnsafeMutableRawPointer?
-                ) -> Void
-            ) = { _, error, _, opaquePointer in
-
-                guard let opaquePointer = opaquePointer else {
-                    fatalError("Could not resolve reference to catpured Swift callback instance")
-                }
-                let opaque = Unmanaged<CapturedCommitCallback>.fromOpaque(opaquePointer).takeUnretainedValue()
-
-                let actualCallback = opaque.closure
-
-                if error == RD_KAFKA_RESP_ERR_NO_ERROR {
-                    actualCallback(.success(()))
-                } else {
-                    let kafkaError = KafkaError.rdKafkaError(wrapping: error)
-                    actualCallback(.failure(kafkaError))
-                }
-            }
-
             changesList.withListPointer { listPointer in
                 rd_kafka_commit_queue(
                     self.kafkaHandle,
                     listPointer,
-                    consumerQueue,
-                    callbackWrapper,
+                    self.mainQueue,
+                    nil,
                     opaquePointer
                 )
             }

Sources/SwiftKafka/KafkaConsumer.swift

@@ -84,7 +84,7 @@ public final class KafkaConsumer: Sendable, Service {
         let client = try RDKafka.createClient(
             type: .consumer,
             configDictionary: config.dictionary,
-            events: [.log],
+            events: [.log, .fetch, .offsetCommit],
             logger: logger
         )
 
@@ -173,20 +173,27 @@ public final class KafkaConsumer: Sendable, Service {
             let nextAction = self.stateMachine.withLockedValue { $0.nextPollLoopAction() }
             switch nextAction {
             case .pollForAndYieldMessage(let client, let source):
-                do {
-                    if let message = try client.consumerPoll() {
-                        // We do not support back pressure, we can ignore the yield result
-                        _ = source.yield(message)
+                let events = client.eventPoll()
+                for event in events {
+                    switch event {
+                    case .consumerMessages(let result):
+                        switch result {
+                        case .success(let message):
+                            // We do not support back pressure, we can ignore the yield result
+                            _ = source.yield(message)
+                        case .failure(let error):
+                            source.finish()
+                            throw error
+                        }
+                    default:
+                        break // Ignore
                     }
-                } 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()
+                let _ = client.eventPoll()
                 try await Task.sleep(for: self.config.pollInterval)
             case .terminatePollLoop:
                 return

Sources/SwiftKafka/KafkaProducer.swift

@@ -218,6 +218,8 @@ public final class KafkaProducer: Service, Sendable {
                     case .deliveryReport(let results):
                         // Ignore YieldResult as we don't support back pressure in KafkaProducer
                         results.forEach { _ = source?.yield($0) }
+                    default:
+                        break // Ignore
                     }
                 }
                 try await Task.sleep(for: self.config.pollInterval)

Tests/SwiftKafkaTests/KafkaConsumerTests.swift

@@ -0,0 +1,84 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2022 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Logging
+import ServiceLifecycle
+@testable import SwiftKafka
+import XCTest
+
+// For testing locally on Mac, do the following:
+//
+// 1. Install Kafka and Zookeeper using homebrew
+//
+// https://medium.com/@Ankitthakur/apache-kafka-installation-on-mac-using-homebrew-a367cdefd273
+//
+// 2. Start Zookeeper & Kafka Server
+//
+// (Homebrew - Apple Silicon)
+// zookeeper-server-start /opt/homebrew/etc/kafka/zookeeper.properties & kafka-server-start /opt/homebrew/etc/kafka/server.properties
+//
+// (Homebrew - Intel Mac)
+// zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties & kafka-server-start /usr/local/etc/kafka/server.properties
+
+final class KafkaConsumerTests: XCTestCase {
+    func testConsumerLog() async throws {
+        let recorder = LogEventRecorder()
+        let mockLogger = Logger(label: "kafka.test.consumer.log") {
+            _ in MockLogHandler(recorder: recorder)
+        }
+
+        // Set no bootstrap servers to trigger librdkafka configuration warning
+        let config = KafkaConsumerConfiguration(
+            consumptionStrategy: .partition(topic: "some topic", partition: .unassigned),
+            bootstrapServers: [],
+            saslMechanism: .gssapi // This should trigger a configuration error
+        )
+
+        let consumer = try KafkaConsumer(config: config, logger: mockLogger)
+
+        let serviceGroup = ServiceGroup(
+            services: [consumer],
+            configuration: ServiceGroupConfiguration(gracefulShutdownSignals: []),
+            logger: .kafkaTest
+        )
+
+        await withThrowingTaskGroup(of: Void.self) { group in
+            // Run Task
+            group.addTask {
+                try await serviceGroup.run()
+            }
+
+            // Sleep for 1s to let poll loop receive log message
+            try! await Task.sleep(for: .seconds(1))
+
+            // Shutdown the serviceGroup
+            await serviceGroup.triggerGracefulShutdown()
+        }
+
+        let recordedEvents = recorder.recordedEvents
+        XCTAssertEqual(1, recordedEvents.count)
+
+        let expectedMessage = """
+        [thrd:app]: Configuration property `sasl.mechanism` set to `GSSAPI` but `security.protocol` \
+        is not configured for SASL: recommend setting `security.protocol` to SASL_SSL or SASL_PLAINTEXT
+        """
+        let expectedLevel = Logger.Level.warning
+        let expectedSource = "CONFWARN"
+
+        let receivedEvent = try XCTUnwrap(recordedEvents.first, "Expected log event, but found none")
+        XCTAssertEqual(expectedMessage, receivedEvent.message.description)
+        XCTAssertEqual(expectedLevel, receivedEvent.level)
+        XCTAssertEqual(expectedSource, receivedEvent.source)
+    }
+}

Tests/SwiftKafkaTests/KafkaProducerTests.swift

@@ -13,7 +13,6 @@
 //===----------------------------------------------------------------------===//
 
 import Logging
-import NIOConcurrencyHelpers
 import NIOCore
 import ServiceLifecycle
 @testable import SwiftKafka
@@ -315,80 +314,4 @@ final class KafkaProducerTests: XCTestCase {
 
         XCTAssertNil(producerCopy)
     }
-
-    // MARK: - Mocks
-
-    internal struct LogEvent {
-        let level: Logger.Level
-        let message: Logger.Message
-        let source: String
-    }
-
-    internal struct LogEventRecorder {
-        let _recordedEvents = NIOLockedValueBox<[LogEvent]>([])
-
-        var recordedEvents: [LogEvent] {
-            self._recordedEvents.withLockedValue { $0 }
-        }
-
-        func record(_ event: LogEvent) {
-            self._recordedEvents.withLockedValue { $0.append(event) }
-        }
-    }
-
-    internal struct MockLogHandler: LogHandler {
-        let recorder: LogEventRecorder
-
-        init(recorder: LogEventRecorder) {
-            self.recorder = recorder
-        }
-
-        func log(
-            level: Logger.Level,
-            message: Logger.Message,
-            metadata: Logger.Metadata?,
-            source: String,
-            file: String,
-            function: String,
-            line: UInt
-        ) {
-            self.recorder.record(LogEvent(level: level, message: message, source: source))
-        }
-
-        private var _logLevel: Logger.Level?
-        var logLevel: Logger.Level {
-            get {
-                // get from config unless set
-                return self._logLevel ?? .debug
-            }
-            set {
-                self._logLevel = newValue
-            }
-        }
-
-        private var _metadataSet = false
-        private var _metadata = Logger.Metadata() {
-            didSet {
-                self._metadataSet = true
-            }
-        }
-
-        public var metadata: Logger.Metadata {
-            get {
-                return self._metadata
-            }
-            set {
-                self._metadata = newValue
-            }
-        }
-
-        subscript(metadataKey metadataKey: Logger.Metadata.Key) -> Logger.Metadata.Value? {
-            get {
-                return self._metadata[metadataKey]
-            }
-            set {
-                self._metadata[metadataKey] = newValue
-            }
-        }
-    }
 }