Have two factory methods creating KafkaProducer

felixschlegel · felixschlegel · commit ff2cb71c46fd · 2023-06-14T14:02:58.000+01:00
Motivation:

We want to have a `KafkaProducer` that is not consuming any
acknowledgements. This means it is initialized without a
`deliveryReportCallback` which in turn means that `librdkafka` will not
queue any incoming acknowledgements which prevents us from running out
of memory in that case.

Modifications:

* add two new factory methods for creating `KafkaProducer`:
    * `KafkaProducer.newProducer`
    * `KafkaProducer.newProducerWithAcknowledgements`
* update README
diff --git a/README.md b/README.md
@@ -11,7 +11,7 @@ The `sendAsync(_:)` method of `KafkaProducer` returns a message-id that can late
 ```swift
 let config = KafkaProducerConfig(bootstrapServers: ["localhost:9092"])
 
-let (producer, acknowledgements) = try await KafkaProducer.newProducer(
+let (producer, acknowledgements) = try await KafkaProducer.newProducerWithAcknowledgements(
     config: config,
     logger: .kafkaTest // Your logger here
 )
diff --git a/Sources/SwiftKafka/KafkaPollingSystem.swift b/Sources/SwiftKafka/KafkaPollingSystem.swift
@@ -33,15 +33,15 @@ final class KafkaPollingSystem<Element>: Sendable {
         self.stateMachine = NIOLockedValueBox(StateMachine())
     }
 
-    /// Initialize the ``KafkaPollingSystem`` and create the ``KafkaAsyncSequence`` that publishes
+    /// Initialize the ``KafkaPollingSystem`` and create the ``NIOAsyncSequenceProducer`` that publishes
     /// message acknowledgements.
     ///
     /// We use this second `initialize()` method to support delayed initialization,
     /// which is needed because the initialization ``NIOAsyncSequenceProducer`` requires a reference
     /// to an existing ``KafkaPollingSystem`` object but our ``StateMachine`` in turn needs a reference to
     /// the ``NIOAsyncSequenceProducer.Source`` object.
     ///
-    /// - Returns: The newly created ``KafkaAsyncSequence`` object.
+    /// - Returns: The newly created ``NIOAsyncSequenceProducer`` object.
     func initialize(
         backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark,
         pollClosure: @escaping () -> Void
diff --git a/Sources/SwiftKafka/KafkaProducer.swift b/Sources/SwiftKafka/KafkaProducer.swift
@@ -70,58 +70,91 @@ public actor KafkaProducer {
     /// Mechanism that polls the Kafka cluster for updates periodically.
     private let pollingSystem: KafkaPollingSystem<
         Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>
-    >
+    >?
     /// Used for handling the connection to the Kafka cluster.
     private let client: KafkaClient
 
-    // Private. `KafkaProducer.newProducer` should be used to create a new producer.
+    // Private initializer, use factory methods to create KafkaProducer
     /// Initialize a new ``KafkaProducer``.
     /// - Parameter config: The ``KafkaProducerConfig`` for configuring the ``KafkaProducer``.
     /// - Parameter topicConfig: The ``KafkaTopicConfig`` used for newly created topics.
     /// - Parameter logger: A logger.
     /// - Throws: A ``KafkaError`` if initializing the producer failed.
     private init(
-        config: KafkaProducerConfig,
+        client: KafkaClient,
+        pollingSystem: KafkaPollingSystem<Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>>? = nil,
         topicConfig: KafkaTopicConfig,
         logger: Logger
     ) async throws {
+        self.client = client
+        self.pollingSystem = pollingSystem
         self.topicConfig = topicConfig
-        self.logger = logger
         self.topicHandles = [:]
+        self.logger = logger
         self.state = .started
+    }
 
-        self.pollingSystem = KafkaPollingSystem<Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>>()
-
-        self.client = try RDKafka.createClient(
+    /// Initialize a new ``KafkaProducer`` that ignores incoming message acknowledgements.
+    /// - Parameter config: The ``KafkaProducerConfig`` for configuring the ``KafkaProducer``.
+    /// - Parameter topicConfig: The ``KafkaTopicConfig`` used for newly created topics.
+    /// - Parameter logger: A logger.
+    /// - Returns: The newly created ``KafkaProducer``.
+    /// - Throws: A ``KafkaError`` if initializing the producer failed.
+    public static func newProducer(
+        config: KafkaProducerConfig = KafkaProducerConfig(),
+        topicConfig: KafkaTopicConfig = KafkaTopicConfig(),
+        logger: Logger
+    ) async throws -> KafkaProducer {
+        let client = try RDKafka.createClient(
             type: .producer,
             configDictionary: config.dictionary,
-            callback: { [logger, pollingSystem] messageResult in
-                guard let messageResult else {
-                    logger.error("Could not resolve acknowledged message")
-                    return
-                }
+            // Having no callback will discard any incoming acknowlegement messages
+            // Ref: rdkafka_broker.c:rd_kafka_dr_msgq
+            callback: nil,
+            logger: logger
+        )
 
-                pollingSystem.yield(messageResult)
-            },
-            logger: self.logger
+        let producer = try await KafkaProducer(
+            client: client,
+            pollingSystem: nil, // we don't receive acknowlegements so no pollingSystem needed
+            topicConfig: topicConfig,
+            logger: logger
         )
+
+        return producer
     }
 
-    /// Initialize a new ``KafkaProducer`` alongside a ``KafkaAsyncSequence`` that can be used
+    /// Initialize a new ``KafkaProducer`` alongside a ``KafkaMessageAcknowledgements`` `AsyncSequence` that can be used
     /// to receive message acknowlegements.
     /// - Parameter config: The ``KafkaProducerConfig`` for configuring the ``KafkaProducer``.
     /// - Parameter topicConfig: The ``KafkaTopicConfig`` used for newly created topics.
     /// - Parameter logger: A logger.
-    /// - Returns: A tuple containing the created ``KafkaProducer`` and the ``KafkaAsyncSequence``
-    /// used for receiving message acknowledgements.
+    /// - Returns: A tuple containing the created ``KafkaProducer`` and the ``KafkaMessageAcknowledgements``
+    /// `AsyncSequence` used for receiving message acknowledgements.
     /// - Throws: A ``KafkaError`` if initializing the producer failed.
-    public static func newProducer(
+    public static func newProducerWithAcknowledgements(
         config: KafkaProducerConfig = KafkaProducerConfig(),
         topicConfig: KafkaTopicConfig = KafkaTopicConfig(),
         logger: Logger
-    ) async throws -> (KafkaProducer, KafkaAsyncSequence<Acknowledgement>) {
+    ) async throws -> (KafkaProducer, KafkaMessageAcknowledgements) {
+        let pollingSystem = KafkaPollingSystem<Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>>()
+        let client = try RDKafka.createClient(
+            type: .producer,
+            configDictionary: config.dictionary,
+            callback: { [logger, pollingSystem] messageResult in
+                guard let messageResult else {
+                    logger.error("Could not resolve acknowledged message")
+                    return
+                }
+
+                pollingSystem.yield(messageResult)
+            },
+            logger: logger
+        )
+
         let producer = try await KafkaProducer(
-            config: config,
+            client: client,
+            pollingSystem: pollingSystem,
             topicConfig: topicConfig,
             logger: logger
         )
@@ -132,8 +165,7 @@ public actor KafkaProducer {
             highWatermark: 50
         )
 
-        let client = producer.client
-        let sequence = producer.pollingSystem.initialize(
+        let _sequence = pollingSystem.initialize(
             backPressureStrategy: backPressureStrategy,
             pollClosure: { [client] in
                 client.withKafkaHandlePointer { handle in
@@ -142,8 +174,9 @@ public actor KafkaProducer {
                 return
             }
         )
+        let acknowlegementsSequence = KafkaMessageAcknowledgements(wrappedSequence: _sequence)
 
-        return (producer, sequence)
+        return (producer, acknowlegementsSequence)
     }
 
     /// Method to shutdown the ``KafkaProducer``.
@@ -163,15 +196,15 @@ public actor KafkaProducer {
 
     private func _shutDownGracefully(timeout: Int32) async {
         await withCheckedContinuation { (continuation: CheckedContinuation<Void, Never>) in
-            // Wait 10 seconds for outstanding messages to be sent and callbacks to be called
+            // Wait `timeout` seconds for outstanding messages to be sent and callbacks to be called
             self.client.withKafkaHandlePointer { handle in
                 rd_kafka_flush(handle, timeout)
                 continuation.resume()
             }
         }
 
         // Kill poll loop in polling system
-        self.pollingSystem.terminate()
+        self.pollingSystem?.terminate()
 
         for (_, topicHandle) in self.topicHandles {
             rd_kafka_topic_destroy(topicHandle)
@@ -186,7 +219,10 @@ public actor KafkaProducer {
     /// - Returns: An awaitable task representing the execution of the poll loop.
     public func run(pollInterval: Duration = .milliseconds(100)) async throws {
         // TODO(felix): make pollInterval part of config -> easier to adapt to Service protocol (service-lifecycle)
-        try await self.pollingSystem.run(pollInterval: pollInterval)
+        guard let pollingSystem else {
+            fatalError("Method \(#function) should only be used with the acknowledgement receiving producer.")
+        }
+        try await pollingSystem.run(pollInterval: pollInterval)
     }
 
     /// Send messages to the Kafka cluster asynchronously, aka "fire and forget".
diff --git a/Tests/IntegrationTests/SwiftKafkaTests.swift b/Tests/IntegrationTests/SwiftKafkaTests.swift
@@ -70,7 +70,7 @@ final class SwiftKafkaTests: XCTestCase {
 
     func testProduceAndConsumeWithConsumerGroup() async throws {
         let testMessages = Self.createTestMessages(topic: self.uniqueTestTopic, count: 10)
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.producerConfig, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.producerConfig, logger: .kafkaTest)
 
         await withThrowingTaskGroup(of: Void.self) { group in
             // Run Task
@@ -122,7 +122,7 @@ final class SwiftKafkaTests: XCTestCase {
 
     func testProduceAndConsumeWithAssignedTopicPartition() async throws {
         let testMessages = Self.createTestMessages(topic: self.uniqueTestTopic, count: 10)
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.producerConfig, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.producerConfig, logger: .kafkaTest)
 
         await withThrowingTaskGroup(of: Void.self) { group in
             // Run Task
@@ -175,7 +175,7 @@ final class SwiftKafkaTests: XCTestCase {
 
     func testProduceAndConsumeWithCommitSync() async throws {
         let testMessages = Self.createTestMessages(topic: self.uniqueTestTopic, count: 10)
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.producerConfig, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.producerConfig, logger: .kafkaTest)
 
         await withThrowingTaskGroup(of: Void.self) { group in
             // Run Task
@@ -245,7 +245,7 @@ final class SwiftKafkaTests: XCTestCase {
 
     private static func sendAndAcknowledgeMessages(
         producer: KafkaProducer,
-        acknowledgements: KafkaAsyncSequence<KafkaProducer.Acknowledgement>,
+        acknowledgements: KafkaMessageAcknowledgements,
         messages: [KafkaProducerMessage]
     ) async throws {
         var messageIDs = Set<UInt>()
diff --git a/Tests/SwiftKafkaTests/KafkaProducerTests.swift b/Tests/SwiftKafkaTests/KafkaProducerTests.swift
@@ -52,7 +52,7 @@ final class KafkaProducerTests: XCTestCase {
     }
 
     func testSendAsync() async throws {
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.config, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.config, logger: .kafkaTest)
 
         await withThrowingTaskGroup(of: Void.self) { group in
 
@@ -91,7 +91,7 @@ final class KafkaProducerTests: XCTestCase {
     }
 
     func testSendAsyncEmptyMessage() async throws {
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.config, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.config, logger: .kafkaTest)
 
         await withThrowingTaskGroup(of: Void.self) { group in
 
@@ -129,7 +129,7 @@ final class KafkaProducerTests: XCTestCase {
     }
 
     func testSendAsyncTwoTopics() async throws {
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.config, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.config, logger: .kafkaTest)
         await withThrowingTaskGroup(of: Void.self) { group in
 
             // Run Task
@@ -185,7 +185,7 @@ final class KafkaProducerTests: XCTestCase {
     }
 
     func testProducerNotUsableAfterShutdown() async throws {
-        let (producer, acks) = try await KafkaProducer.newProducer(config: self.config, logger: .kafkaTest)
+        let (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.config, logger: .kafkaTest)
         await producer.shutdownGracefully()
 
         await withThrowingTaskGroup(of: Void.self) { group in
@@ -209,7 +209,7 @@ final class KafkaProducerTests: XCTestCase {
 
                 // This subscribes to the acknowledgements stream and immediately terminates the stream.
                 // Required to kill the run task.
-                var iterator: KafkaAsyncSequence<KafkaProducer.Acknowledgement>.AsyncIterator? = acks.makeAsyncIterator()
+                var iterator: KafkaMessageAcknowledgements.AsyncIterator? = acks.makeAsyncIterator()
                 _ = iterator
                 iterator = nil
             }
@@ -218,8 +218,8 @@ final class KafkaProducerTests: XCTestCase {
 
     func testNoMemoryLeakAfterShutdown() async throws {
         var producer: KafkaProducer?
-        var acks: KafkaAsyncSequence<KafkaProducer.Acknowledgement>
-        (producer, acks) = try await KafkaProducer.newProducer(config: self.config, logger: .kafkaTest)
+        var acks: KafkaMessageAcknowledgements
+        (producer, acks) = try await KafkaProducer.newProducerWithAcknowledgements(config: self.config, logger: .kafkaTest)
         _ = acks
 
         weak var producerCopy = producer

Original file line number	Diff line number	Diff line change
@@ -11,7 +11,7 @@ The `sendAsync(_:)` method of `KafkaProducer` returns a message-id that can late
`11`	`11`	```swift
`12`	`12`	`let config = KafkaProducerConfig(bootstrapServers: ["localhost:9092"])`
`13`	`13`
`14`		`-let (producer, acknowledgements) = try await KafkaProducer.newProducer(`
	`14`	`+let (producer, acknowledgements) = try await KafkaProducer.newProducerWithAcknowledgements(`
`15`	`15`	`config: config,`
`16`	`16`	`logger: .kafkaTest // Your logger here`
`17`	`17`	`)`