KafkaProducer: Expose `run()` method #63

felixschlegel · 2023-06-14T15:04:18Z

Modifications

fix typo in SwiftKafkaTests
upgrade minimum OS versions to support the Duration type
move conversion of rd_kafka_message_t to KafkaAcknowledgementResult to RDKafkaConfig so that we can pass the KafkaAcknowledgementResult type as early as possible and don't have to bother with UnsafePointer<rd_kafka_message_t> in all our delivery callback logic
expose KafkaProducer.run() method
README: use TaskGroup in KafkaProducer example
add comments to task groups in README
refactor all tests into using task groups (structured concurrency)

Have two factory methods creating KafkaProducer

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

FranzBusch · 2023-06-15T10:16:19Z

Sources/SwiftKafka/KafkaProducer.swift

-    /// Task that polls the Kafka cluster for updates periodically.
-    private var pollTask: Task<Void, Never>!
+    /// Variable that indicates if our poll loop should be running.
+    private var polling = true


We don't need this property. We can use the state for this started indicates that we are polling

Motivation: * Our current implementation of polling `librdkafka` for new message acknowledgements in `KafkaProducer` did not support backpressure. Also, it made use of some `weak` references which we generally want to avoid due to performance reasons. Modifications: * fix typo in `SwiftKafkaTests` * upgrade minimum OS versions to support the `Duration` type * move conversion of `rd_kafka_message_t` to `KafkaAcknowledgementResult` to `RDKafkaConfig` so that we can pass the `KafkaAcknowledgementResult` type as early as possible and don't have to bother with `UnsafePointer<rd_kafka_message_t>` in all our delivery callback logic * expose KafkaProducer.run() method * `README`: use `TaskGroup` in `KafkaProducer` example * add comments to task groups in `README` * refactor all tests into using task groups (structured concurrency) Have two factory methods creating KafkaProducer 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

FranzBusch · 2023-06-15T12:04:09Z

Sources/SwiftKafka/KafkaProducer.swift

-    public nonisolated let acknowledgements: AcknowledgedMessagesAsyncSequence
-    nonisolated let acknowlegdementsSource: AcknowledgedMessagesAsyncSequence.WrappedSequence.Source
-    private typealias Acknowledgement = Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>
+    private let client: KafkaClient


Can we create a ticket to try to internalise all of the properties here into our state where it makes sense

FranzBusch · 2023-06-15T12:04:20Z

Sources/SwiftKafka/KafkaProducer.swift

+    /// - Parameter logger: A logger.
+    /// - Returns: The newly created ``KafkaProducer``.
+    /// - Throws: A ``KafkaError`` if initializing the producer failed.
+    public static func newProducer(


Suggested change

public static func newProducer(

public static func makeProducer(

FranzBusch · 2023-06-15T12:06:02Z

Sources/SwiftKafka/KafkaProducer.swift

+    /// 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.


Suggested change

/// 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.

/// Initialize a new ``KafkaProducer``.

///

/// This factory method creates a producer without message acknowledgements.

///

/// - Parameter configuration: The ``KafkaProducerConfig`` for configuring the ``KafkaProducer``.

/// - Parameter topicConfiguration: The ``KafkaTopicConfig`` used for newly created topics.

/// - Parameter logger: A logger.

/// - Returns: The newly created ``KafkaProducer``.

/// - Throws: A ``KafkaError`` if initializing the producer failed.

Can we also create a radar to rename the config structs to spell out configuration.

FranzBusch · 2023-06-15T12:07:46Z

Sources/SwiftKafka/KafkaProducer.swift

+    /// 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 ``KafkaMessageAcknowledgements``
+    /// `AsyncSequence` used for receiving message acknowledgements.
+    /// - Throws: A ``KafkaError`` if initializing the producer failed.


Suggested change

/// 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 ``KafkaMessageAcknowledgements``

/// `AsyncSequence` used for receiving message acknowledgements.

/// - Throws: A ``KafkaError`` if initializing the producer failed.

/// Initialize a new ``KafkaProducer`` and a ``KafkaMessageAcknowledgements`` asynchronous sequence.

///

/// Use the asynchronous sequence to consume message acknowledgements.

///

/// - Important: When the asynchronous sequence is deinited the producer will be shutdown.

///

/// - 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 ``KafkaMessageAcknowledgements``

/// `AsyncSequence` used for receiving message acknowledgements.

/// - Throws: A ``KafkaError`` if initializing the producer failed.

FranzBusch · 2023-06-15T12:08:35Z

Sources/SwiftKafka/KafkaProducer.swift

+                    return
+                }
+
+                _ = source.yield(messageResult) // Ignore YieldResult


Let's add why we ignore this result i.e. due to no back pressure support.

FranzBusch · 2023-06-15T12:09:25Z

Sources/SwiftKafka/KafkaProducer.swift

+            self.client.withKafkaHandlePointer { handle in
+                rd_kafka_poll(handle, 0)
+            }


Can we make this a method on the KafkaClient which does the withHandle stuff?

Modifications: * rename `KafkaProducer.newProducer*` to `KafkaProducer.makeProducer*` * update docc documentation for `KafkaProducer.newProducer*` methods * create new method `KafkaClient.poll(timeout:)`

Motivation: Given that we don't support backpressure for our `KafkaProducer`, we can replace the `NIOAsyncSequence` with a `AsyncStream` for the acknowledgements. Furthermore we want to shut down the producer once our stream has terminated. Modifications: * `KafkaProducer`: replace `NIOAsyncSequence` with `AsyncStream` * `KafkaProducerTests.testNoMemoryLeakAfterShutdown` make sure to kill stream otherwise we have a memory leak

felixschlegel force-pushed the fs-kafka-producer-run-method branch 2 times, most recently from 70f32cd to 9fd97e2 Compare June 14, 2023 15:17

FranzBusch reviewed Jun 15, 2023

View reviewed changes

felixschlegel force-pushed the fs-kafka-producer-run-method branch from 9fd97e2 to 28c9e6f Compare June 15, 2023 12:02

felixschlegel requested a review from FranzBusch June 15, 2023 12:03

FranzBusch requested changes Jun 15, 2023

View reviewed changes

felixschlegel added 2 commits June 15, 2023 14:04

Review Franz

3a6d545

Modifications: * rename `KafkaProducer.newProducer*` to `KafkaProducer.makeProducer*` * update docc documentation for `KafkaProducer.newProducer*` methods * create new method `KafkaClient.poll(timeout:)`

felixschlegel force-pushed the fs-kafka-producer-run-method branch from 66c57ef to 84a67d5 Compare June 15, 2023 16:36

felixschlegel mentioned this pull request Jun 15, 2023

Internalise properties of KafkaProducer into its state #64

Open

felixschlegel requested a review from FranzBusch June 16, 2023 10:05

FranzBusch approved these changes Jun 19, 2023

View reviewed changes

FranzBusch merged commit 840242c into swift-server:main Jun 19, 2023

FranzBusch deleted the fs-kafka-producer-run-method branch June 19, 2023 12:02

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KafkaProducer: Expose `run()` method #63

KafkaProducer: Expose `run()` method #63

felixschlegel commented Jun 14, 2023

FranzBusch Jun 15, 2023

FranzBusch Jun 15, 2023

felixschlegel Jun 15, 2023

FranzBusch Jun 15, 2023

FranzBusch Jun 15, 2023

felixschlegel Jun 15, 2023

FranzBusch Jun 15, 2023

FranzBusch Jun 15, 2023

FranzBusch Jun 15, 2023

	public static func newProducer(
	public static func makeProducer(

KafkaProducer: Expose run() method #63

KafkaProducer: Expose run() method #63

Conversation

felixschlegel commented Jun 14, 2023

Modifications

Have two factory methods creating KafkaProducer

Motivation

Modifications

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KafkaProducer: Expose `run()` method #63

KafkaProducer: Expose `run()` method #63