Merge remote-tracking branch 'origin/main' into feature/sc-1976/gsoc-expose-librdkafka-statistics

Author: blindspotbounty

Diff for: Sources/SwiftKafka/Configuration/KafkaProducerConfiguration.swift

@@ -25,6 +25,17 @@ public struct KafkaProducerConfiguration {
     /// >= 1ms - statistics provided every specified interval
     public var statisticsInterval: Duration = .zero
 
+    /// Maximum timeout for flushing outstanding produce requests when the ``KakfaProducer`` is shutting down.
+    /// Default: `10000`
+    public var flushTimeoutMilliseconds: Int = 10000 {
+        didSet {
+            precondition(
+                0...Int(Int32.max) ~= self.flushTimeoutMilliseconds,
+                "Flush timeout outside of valid range \(0...Int32.max)"
+            )
+        }
+    }
+
     // MARK: - Producer-specific Config Properties
 
     /// When set to true, the producer will ensure that messages are successfully produced exactly once and in the original produce order. The following configuration properties are adjusted automatically (if not modified by the user) when idempotence is enabled: max.in.flight.requests.per.connection=5 (must be less than or equal to 5), retries=INT32_MAX (must be greater than 0), acks=all, queuing.strategy=fifo. Producer instantation will fail if user-supplied configuration is incompatible.

Diff for: Sources/SwiftKafka/KafkaConsumer.swift

@@ -12,7 +12,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-import Crdkafka
 import Logging
 import NIOConcurrencyHelpers
 import NIOCore
@@ -128,7 +127,7 @@ public final class KafkaConsumer: Sendable, Service {
         self.config = config
         self.logger = logger
 
-        let client = try RDKafka.createClient(
+        let client = try RDKafkaClient.makeClient(
             type: .consumer,
             configDictionary: config.dictionary,
             events: [.log, .fetch, .offsetCommit],
@@ -164,8 +163,8 @@ public final class KafkaConsumer: Sendable, Service {
             )
         }
 
-        // Events that would be triggered by ``KafkaClient/poll(timeout:)``
-        // are now triggered by ``KafkaClient/consumerPoll``.
+        // Events that would be triggered by ``RDKafkaClient/poll(timeout:)``
+        // are now triggered by ``RDKafkaClient/consumerPoll``.
         try client.pollSetConsumer()
 
         switch config.consumptionStrategy._internal {
@@ -302,7 +301,7 @@ public final class KafkaConsumer: Sendable, Service {
     }
 
     private func _triggerGracefulShutdown(
-        client: KafkaClient,
+        client: RDKafkaClient,
         logger: Logger
     ) {
         do {
@@ -336,7 +335,7 @@ extension KafkaConsumer {
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case initializing(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source,
                 statisticsSource: StatisticsProducer.Source?
             )
@@ -346,15 +345,15 @@ extension KafkaConsumer {
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             /// - Parameter statisticsSource: ``NIOAsyncSequenceProducer/Source`` used for yielding new statistics elements.
             case consuming(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source?,
                 statisticsSource: StatisticsProducer.Source?
             )
             /// The ``KafkaConsumer/triggerGracefulShutdown()`` has been invoked.
             /// We are now in the process of commiting our last state to the broker.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case finishing(client: KafkaClient)
+            case finishing(client: RDKafkaClient)
             /// The ``KafkaConsumer`` is closed.
             case finished
         }
@@ -365,7 +364,7 @@ extension KafkaConsumer {
         /// Delayed initialization of `StateMachine` as the `source` and the `pollClosure` are
         /// not yet available when the normal initialization occurs.
         mutating func initialize(
-            client: KafkaClient,
+            client: RDKafkaClient,
             source: Producer.Source,
             statisticsSource: StatisticsProducer.Source?
         ) {
@@ -387,7 +386,7 @@ extension KafkaConsumer {
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             /// - Parameter statisticsSource: ``NIOAsyncSequenceProducer/Source`` used for yielding new statistics elements.
             case pollForAndYieldMessage(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source?,
                 statisticsSource: StatisticsProducer.Source?
             )
@@ -423,7 +422,7 @@ extension KafkaConsumer {
         enum SetUpConnectionAction {
             /// Set up the connection through ``subscribe()`` or ``assign()``.
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case setUpConnection(client: KafkaClient)
+            case setUpConnection(client: RDKafkaClient)
         }
 
         /// Get action to be taken when wanting to set up the connection through ``subscribe()`` or ``assign()``.
@@ -482,7 +481,7 @@ extension KafkaConsumer {
         enum StoreOffsetAction {
             /// Store the message offset with the given `client`.
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case storeOffset(client: KafkaClient)
+            case storeOffset(client: RDKafkaClient)
         }
 
         /// Get action to take when wanting to store a message offset (to be auto-committed by `librdkafka`).
@@ -508,7 +507,7 @@ extension KafkaConsumer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             case commitSync(
-                client: KafkaClient
+                client: RDKafkaClient
             )
             /// Throw an error. The ``KafkaConsumer`` is closed.
             case throwClosedError
@@ -540,14 +539,14 @@ extension KafkaConsumer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             case triggerGracefulShutdown(
-                client: KafkaClient
+                client: RDKafkaClient
             )
             /// Shut down the ``KafkaConsumer`` and finish the given `source` object.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             case triggerGracefulShutdownAndFinishSource(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 source: Producer.Source,
                 statisticsSource: StatisticsProducer.Source?
             )

Diff for: Sources/SwiftKafka/KafkaProducer.swift

@@ -12,7 +12,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-import Crdkafka
 import Logging
 import NIOConcurrencyHelpers
 import NIOCore
@@ -136,7 +135,7 @@ public final class KafkaProducer: Service, Sendable {
             }
         }
 
-        let client = try RDKafka.createClient(
+        let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: config.dictionary,
             events: [.log], // No .deliveryReport here!
@@ -189,7 +188,7 @@ public final class KafkaProducer: Service, Sendable {
         )
         let source = sourceAndSequence.source
 
-        let client = try RDKafka.createClient(
+        let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: config.dictionary,
             events: [.log, .deliveryReport],
@@ -268,9 +267,15 @@ public final class KafkaProducer: Service, Sendable {
                 if !flushing || events.isEmpty {
                     try await Task.sleep(for: self.config.pollInterval)
                 }
-            case .terminatePollLoopAndFinishSource(let source, let statisticsSource):
+            case .flushFinishSourceAndTerminatePollLoop(let client, let source, let statisticsSource):
+                precondition(
+                    0...Int(Int32.max) ~= self.config.flushTimeoutMilliseconds,
+                    "Flush timeout outside of valid range \(0...Int32.max)"
+                )
+                try await client.flush(timeoutMilliseconds: Int32(self.config.flushTimeoutMilliseconds))
                 source?.finish()
                 statisticsSource?.finish()
+                statisticsSource?.finish()
                 return
             case .terminatePollLoop:
                 return
@@ -329,7 +334,7 @@ extension KafkaProducer {
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
             /// - Parameter topicHandles: Class containing all topic names with their respective `rd_kafka_topic_t` pointer.
             case started(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 messageIDCounter: UInt,
                 source: Producer.Source?,
                 statisticsSource: StatisticsProducer.Source?,
@@ -340,8 +345,8 @@ extension KafkaProducer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
-            case flushing(
-                client: KafkaClient,
+            case finishing(
+                client: RDKafkaClient,
                 source: Producer.Source?,
                 statisticsSource: StatisticsProducer.Source?
             )
@@ -355,7 +360,7 @@ extension KafkaProducer {
         /// Delayed initialization of `StateMachine` as the `source` is not yet available
         /// when the normal initialization occurs.
         mutating func initialize(
-            client: KafkaClient,
+            client: RDKafkaClient,
             source: Producer.Source?,
             statisticsSource: StatisticsProducer.Source?
         ) {
@@ -377,11 +382,12 @@ extension KafkaProducer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
-            case pollAndYield(client: KafkaClient, source: Producer.Source?, statisticsSource: StatisticsProducer.Source?, flushing: Bool = false)
+            case pollAndYield(client: RDKafkaClient, source: Producer.Source?, statisticsSource: StatisticsProducer.Source?, flushing: Bool = false)
             /// Terminate the poll loop and finish the given `NIOAsyncSequenceProducerSource`.
             ///
+            /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
-            case terminatePollLoopAndFinishSource(source: Producer.Source?, statisticsSource: StatisticsProducer.Source?)
+            case flushFinishSourceAndTerminatePollLoop(client: RDKafkaClient, source: Producer.Source?, statisticsSource: StatisticsProducer.Source?)
             /// Terminate the poll loop.
             case terminatePollLoop
         }
@@ -396,13 +402,8 @@ extension KafkaProducer {
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .started(let client, _, let source, let statisticsSource, _):
                 return .pollAndYield(client: client, source: source, statisticsSource: statisticsSource)
-            case .flushing(let client, let source, let statisticsSource):
-                if client.outgoingQueueSize > 0 {
-                    return .pollAndYield(client: client, source: source, statisticsSource: statisticsSource, flushing: true)
-                } else {
-                    self.state = .finished
-                    return .terminatePollLoopAndFinishSource(source: source, statisticsSource: statisticsSource)
-                }
+            case .finishing(let client, let source, let statisticsSource):
+                return .flushFinishSourceAndTerminatePollLoop(client: client, source: source, statisticsSource: statisticsSource)
             case .finished:
                 return .terminatePollLoop
             }
@@ -414,7 +415,7 @@ extension KafkaProducer {
             ///
             /// - Important: `newMessageID` is the new message ID assigned to the message to be sent.
             case send(
-                client: KafkaClient,
+                client: RDKafkaClient,
                 newMessageID: UInt,
                 topicHandles: RDKafkaTopicHandles
             )
@@ -444,8 +445,8 @@ extension KafkaProducer {
                     newMessageID: newMessageID,
                     topicHandles: topicHandles
                 )
-            case .flushing:
-                throw KafkaError.connectionClosed(reason: "Producer in the process of flushing and shutting down")
+            case .finishing:
+                throw KafkaError.connectionClosed(reason: "Producer in the process of finishing")
             case .finished:
                 throw KafkaError.connectionClosed(reason: "Tried to produce a message with a closed producer")
             }
@@ -479,9 +480,9 @@ extension KafkaProducer {
                 }
                 self.state = .started(client: client, messageIDCounter: counter, source: nil, statisticsSource: statisticsSource, topicHandles: topicHandlers)
                 return .finishSource(source: source)
-            case .flushing(let client, let source, let statisticsSource):
+            case .finishing(let client, let source, let statisticsSource):
                 // Setting source to nil to prevent incoming acknowledgements from buffering in `source`
-                self.state = .flushing(client: client, source: nil, statisticsSource: statisticsSource)
+                self.state = .finishing(client: client, source: nil, statisticsSource: statisticsSource)
                 return .finishSource(source: source)
             case .finished:
                 break
@@ -498,13 +499,10 @@ extension KafkaProducer {
                     fatalError("stopStatistics() must not be invoked more than once")
                 }
                 self.state = .started(client: client, messageIDCounter: counter, source: source, statisticsSource: nil, topicHandles: topicHandlers)
-                client.withKafkaHandlePointer { kafkaHandle in
-                    rd_kafka_conf_set_stats_cb(<#T##conf: OpaquePointer!##OpaquePointer!#>, <#T##stats_cb: ((OpaquePointer?, UnsafeMutablePointer<CChar>?, Int, UnsafeMutableRawPointer?) -> Int32)!##((OpaquePointer?, UnsafeMutablePointer<CChar>?, Int, UnsafeMutableRawPointer?) -> Int32)!##(OpaquePointer?, UnsafeMutablePointer<CChar>?, Int, UnsafeMutableRawPointer?) -> Int32#>)
-                }
                 return .finishStatisticsSource(statisticsSource: statisticsSource)
-            case .flushing(let client, let source, let statisticsSource):
+            case .finishing(let client, let source, let statisticsSource):
                 // Setting source to nil to prevent incoming acknowledgements from buffering in `source`
-                self.state = .flushing(client: client, source: source, statisticsSource: nil)
+                self.state = .finishing(client: client, source: source, statisticsSource: nil)
                 return .finishStatisticsSource(statisticsSource: statisticsSource)
             case .finished:
                 break
@@ -520,8 +518,8 @@ extension KafkaProducer {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .started(let client, _, let source, let statisticsSource, _):
-                self.state = .flushing(client: client, source: source, statisticsSource: statisticsSource)
-            case .flushing, .finished:
+                self.state = .finishing(client: client, source: source, statisticsSource: statisticsSource)
+            case .finishing, .finished:
                 break
             }
         }