KafkaProducer: Backpressure for Ack'd messages #53

felixschlegel · 2023-05-26T15:28:40Z

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

create a new class called KafkaBackPressurePollingSystem
implement backpressure for KafkaProducer's poll-loop with KafkaBackPressurePollingSystem
fix typo in SwiftKafkaTests

TODO

write sensible tests for this backpressure mechanism and the state machine
how to handle dependency injection into our KafkaBackPressurePollingSystem, given that some of its dependencies are not Sendable
discuss if we want to expose a Kafka.run() method to the user or if we want to keep the poll task to ourselves (cc: @FranzBusch )

FranzBusch · 2023-05-30T09:31:56Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    ///
+    /// - Parameter pollIntervalNanos: The poll interval in nanoseconds.
+    /// - Returns: An awaitable task representing the execution of the poll loop.
+    func run(pollIntervalNanos: UInt64) async {


Can we use a Duration here instead?

Yeah works, will have to update our minimum Package version to macOS .v13 / iOS .v16 though 👍

FranzBusch · 2023-05-30T09:32:46Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    typealias Producer = NIOAsyncSequenceProducer<Element, NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark, KafkaBackPressurePollingSystem>
+
+    /// The state machine that manages the system's state transitions.
+    let stateMachine: StateMachine


This is the main thing that we have to protect via a lock. Can you put this inside a NIOLockedValueBox please

FranzBusch · 2023-05-30T09:33:42Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    private let _client: NIOLockedValueBox<KafkaClient?>
+    private let _sequenceSource: NIOLockedValueBox<Producer.Source?>


Rather than having state outside of the state machine just put it into the associated values of the state. This way they are protected by the lock. If an action has to do something with those just return it in the action.

FranzBusch · 2023-05-30T09:34:36Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+            let action = self.stateMachine.nextPollLoopAction()
+
+            switch action {
+            case .poll:


Can we name this action poolAndSleep

FranzBusch · 2023-05-30T09:35:16Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+                self.client?.withKafkaHandlePointer { handle in
+                    rd_kafka_poll(handle, 0)
+                }
+                try? await Task.sleep(nanoseconds: pollIntervalNanos)


We shouldn't silence the error here. If our task got cancelled we should exit out of the run loop. Importantly though we probably want to finish the source then.

FranzBusch · 2023-05-30T09:36:59Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+            /// of `produceMore()` to continue producing messages.
+            case loopSuspended(CheckedContinuation<Void, Never>)
+            /// The system is shut down.
+            case shutDown


Can we rename this to finished to follow our other state machine naming

FranzBusch · 2023-05-30T09:37:45Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+        }
+
+        /// The current state of the state machine.
+        let state = NIOLockedValueBox(State.initial)


Using the locked value box here around an enum has not effect. The enum itself is thread safe. What is not thread safe are the calls to the various mutating methods on the StateMachine.

FranzBusch · 2023-05-30T09:38:20Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    /// The state machine used by the ``KafkaBackPressurePollingSystem``.
+    struct StateMachine {
+        /// The possible actions for the poll loop.
+        enum PollLoopAction {


Can we move the actions right above the method that returns them. This makes it easier to understand what is going on.

FranzBusch · 2023-05-30T09:39:24Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+        /// Triggers an event in the state machine, causing a state transition.
+        ///
+        /// - Parameter event: The event to be triggered.
+        func eventTriggered(_ event: Event) {


Event triggered is way too broad. Please introduce separate methods for all the various events you model here e.g. pollSuspended, pollCancelled etc.

FranzBusch · 2023-05-30T09:40:23Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+import NIOConcurrencyHelpers
+import NIOCore
+
+// TODO: write test for evaluating right state transitions


To make this testable instead of passing in a KafkaClient you can pass in a closure that polls. In real code you can pass in a closure that does the actual polling and in the test you can pass something in which you can assert on

felixschlegel · 2023-05-31T13:50:55Z

Thanks for your review! I have implemented your changes and added 3 test cases 😄
I still get warnings about conforming to Sendable. Please see my comments inline!

felixschlegel · 2023-05-31T13:55:11Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    /// The state machine used by the ``KafkaBackPressurePollingSystem``.
+    struct StateMachine: Sendable {
+        /// The ``NIOAsyncSequenceProducer.Source`` used for yielding the messages to the ``NIOAsyncSequenceProducer``.
+        var sequenceSource: Producer.Source? // TODO: make sendable?


NIOAsyncSequenceProducer.Source is not Sendable at the moment which is giving us a warning here. Should I try making it Sendable in swift-nio or do we have a workaround for that in our case here?

The StateMachine doesn't necessarily need to be Sendable we are holding it in a Lock which makes it Sendable We just need to use @unchecked Sendable on the KafkaBackPressurePollingSystem

felixschlegel · 2023-05-31T13:56:34Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    let stateMachineLock: NIOLockedValueBox<StateMachine>
+
+    /// Closure that takes care of polling `librdkafka` for new messages.
+    var pollClosure: (() -> Void)?


Atm this is a var as it captures self in KafkaProducer so it can get set after our KafkaProducer has been initialized. However, having a var here makes us not conform to Sendable.

We should move this closure into the state machine. Furthermore, I don't see why this should capture the KafkaProducer the only thing that we need is capturing the KafkaClient to call poll on it

FranzBusch

We are getting close! The state machine looks already quite good we just have to move some of the things outside of the state machine. I left some comments inline

FranzBusch · 2023-06-01T10:00:42Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    let stateMachineLock: NIOLockedValueBox<StateMachine>
+
+    /// Closure that takes care of polling `librdkafka` for new messages.
+    var pollClosure: (() -> Void)?


We should move this closure into the state machine. Furthermore, I don't see why this should capture the KafkaProducer the only thing that we need is capturing the KafkaClient to call poll on it

FranzBusch · 2023-06-01T10:01:10Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    /// Closure that takes care of polling `librdkafka` for new messages.
+    var pollClosure: (() -> Void)?
+    /// The ``NIOAsyncSequenceProducer.Source`` used for yielding the messages to the ``NIOAsyncSequenceProducer``.
+    var sequenceSource: Producer.Source? {


Can't we pass this in the init or rather could we actually create the sequence in here and return it from a factory method?

FranzBusch · 2023-06-01T10:01:36Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    /// - Parameter pollInterval: The desired time interval between two consecutive polls.
+    /// - Returns: An awaitable task representing the execution of the poll loop.
+    func run(pollInterval: Duration) async {
+        actionLoop: while true {


We don't need the label here instead we can just use return instead of break further down

FranzBusch · 2023-06-01T10:02:12Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+                do {
+                    try await Task.sleep(for: pollInterval)
+                } catch {
+                    self.stateMachineLock.withLockedValue { $0.shutDown() }


We probably have to call source.finish() here?

source.finish() gets invoked by shutDown() 👍

Yes but that's inside the state machine and we want to have this outside the state machine! Since the state machine should not execute any side effects

FranzBusch · 2023-06-01T10:03:08Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    }
+
+    /// The delivery report callback function that handles acknowledged messages.
+    private(set) lazy var deliveryReportCallback: (UnsafePointer<rd_kafka_message_t>?) -> Void = { messagePointer in


In the scope of this PR can we make a change so that we don't use UnsafePointer here but rather bridge the call-back to something well typed inside the method where we set the callback?

FranzBusch · 2023-06-01T10:11:08Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+            case .finished:
+                return
+            case .stopProducing(let existingContinuation) where existingContinuation != nil:
+                fatalError("Created leaking continuation")


Suggested change

fatalError("Created leaking continuation")

fatalError("Internal state inconsistency. Run loop is running more than once")

FranzBusch · 2023-06-01T10:11:35Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+
+        /// Shut down the state machine and finish producing elements.
+        mutating func shutDown() {
+            self.sequenceSource?.finish()


Same here please move this outside of the state machine

FranzBusch · 2023-06-01T10:11:49Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+            case .stopProducing(let continuation):
+                continuation?.resume()
+                fallthrough
+            default:
+                self.sequenceSource?.finish()


Please apply all previous comments to this as well

FranzBusch · 2023-06-01T10:12:00Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+        }
+
+        /// Yields a new elements to the ``NIOAsyncSequenceProducer``.
+        fileprivate func yield(_ element: Element) -> Producer.Source.YieldResult {


Please yield outside of the state machine

FranzBusch · 2023-06-01T10:13:26Z

Sources/SwiftKafka/KafkaBackPressurePollingSystem.swift

+    ///
+    /// - Parameter pollInterval: The desired time interval between two consecutive polls.
+    /// - Returns: An awaitable task representing the execution of the poll loop.
+    func run(pollInterval: Duration) async {


We should make sure the run() method is only called once. We can do this with a method in the state machine that transition from initial to producing or sets a flag on the initial state.

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: * create a new class called `KafkaBackPressurePollingSystem` * implement backpressure for `KafkaProducer`'s poll-loop with `KafkaBackPressurePollingSystem` * fix typo in `SwiftKafkaTests`

Modifications: * upgrade minimum OS versions to support the `Duration` type * change type of `KafkaBackPressurePollingSystem.run`'s `pollInterval` to `Duration` * rename `PollLoopAction.poll` -> `PollLoopAction.pollAndSleep` * rename `StateMachine.State.shutDown` -> `StateMachine.State.finished` * move lock outside of `StateMachine` * synchronize access to KafkaBackPressurePollingSystem.client/producerSource` by moving them into the `StateMachine` * break down `StateMachine.eventTriggered` into separate methods * finish `sequenceSource` when `StateMachine.shutDown` is invoked * shutdown poll loop when `Task.sleep` fails

…o one

Motivation: We want to be able to inject a closure that contains the polling mechanism into our `KafkaBackPressurePollingSystem` so that we can assert that it has been called in testing.

Condition for checking leaking continuations in `KafkaBackPressurePollingSystem.StateMachine.suspendLoop` was wrong.

Modifications: * added test testing that invocations to the `pollClosure` in `KafkaBackPressurePollingSystem` only occur when backpressure allows it * added test that makes sure the `KafkaBackPressurePollingSystem` shuts down properly when it's `run()` method's parent task gets cancelled

Modifications: * remove flakey tests testing that poll is not invoked when loop is suspended * add dedicated test that tests the case mentioned above

Modifications: * moved `KafkaAcknowledgedMessageAsyncSequnce` to `KafkaBackPressurePollingSystem` * made `init` of `KafkaBackPressurePollingSystem` private and make it only instantiable via the `KafkaBackPressurePollingSystem.createSystemAndSequence` factory method

Modifications: * remove `actionLoop` label and replace `break actionLoop` with plain `return`s * remove unused type `KafkaBackPressurePollingSystem.StateMachine.Event` * improve syntax of `.stopProducing` case in `suspendLoop()` * better error message for `.stopProducing` case in `suspendLoop()` * `KafkaBackPressurePollingSystem`: remove `default` and `fallthrough`s in `switch` statements * `fatalError` if `KafkaBackPressurePollingSystem.run` is invoked more than once

Modifications: * create a new enum `StateMachine.Command` * return `StateMachine.Command` on the `StateMachine` transitions with side-effects * execute `StateMachine.Command`s inside a dedicated command handler in `KafkaBackPressurePollingSystem` * remove `KafkaBackPressurePollingSystem.StateMachine.yield(_:)` method

Modifications: * 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

…tem`

Modifications: * add locked `running` variable to `KafkaBackPressurePollingSystem` to determin if `run` has been invoked already (Bug fix, relying on `state` here was wrong as methods other than `run()` can change to state from `.initial` to anything else * made `init` of `KafkaBackPressurePollingSystem` `internal` to avoid having a `NIOAsyncSequenceProducerDelegate` which could make our tests flakey

Motivation: In `KafkaProducer`, instead of launching the `runTask` ourselves, we expose a `run()` method to the user. By doing so, we can adapt to the `Service` protocol from `swift-service-lifecycle` in the future. Modifications: * create a new method `KafkaProducer.run()` wrapping the `pollingSystem`'s `run()` method * make tests use their own `runTask` * adapt `KafkaProducer` example in `README`

Modifications: * remove `ClosureWrapper` classes * make `pollClosure` `Optional` * make `pollClosure` a get/set property of `KafkaPollingSystem` * adjust tests

Modifications: * move common logic into `setUp()` method

FranzBusch · 2023-06-08T10:00:16Z

README.md

+    }
+
+    group.addTask {
+        let messageID = try await producer.sendAsync(


As a follow up we should probably rename the method to send and just document that it is async under the hood since we are not planning to add a sync variant to this type anymore

FranzBusch · 2023-06-08T10:49:46Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+    var source: Producer.Source? {
+        get {
+            self.stateMachine.withLockedValue { $0.source }
+        }
+        set {
+            self.stateMachine.withLockedValue { $0.source = newValue }
+        }
+    }
+
+    /// Closure that is used to poll the upstream producer for new updates.
+    /// In our case the upstream producer is the Kafka cluster.
+    var pollClosure: (() -> Void)? {
+        get {
+            self.stateMachine.withLockedValue { $0.pollClosure }
+        }
+        set {
+            self.stateMachine.withLockedValue { $0.pollClosure = newValue }
+        }
+    }


Let's try to inject this via a factory method and init.

private let stateMachine: NIOLockedValueBox<StateMachine?> init() {} func initialize(pollClosure: @escaping () -> Void, source: Producer.Source) { ... } static makePollingSystem() -> Self { }

FranzBusch · 2023-06-08T10:58:39Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+    func run(pollInterval: Duration) async throws {
+        switch self.stateMachine.withLockedValue({ $0.run() }) {
+        case .alreadyClosed:
+            return


We should throw an error here

FranzBusch · 2023-06-08T13:18:32Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+    /// Yield new elements to the underlying `NIOAsyncSequenceProducer`.
+    func yield(_ element: Element) {
+        self.stateMachine.withLockedValue { stateMachine in
+            switch stateMachine.state {


Why are we switching on the state here? We should switch on the state inside the state machine and return an action

FranzBusch · 2023-06-08T13:18:42Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+        }
+
+        /// The current state of the state machine.
+        var state: State


Suggested change

var state: State

private var state: State

FranzBusch · 2023-06-08T13:19:36Z

Sources/SwiftKafka/RDKafka/RDKafkaConfig.swift

+    /// 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``.
+    static func convertMessageToAcknowledgementResult(


Suggested change

static func convertMessageToAcknowledgementResult(

private static func convertMessageToAcknowledgementResult(

FranzBusch · 2023-06-08T13:20:09Z

Tests/IntegrationTests/SwiftKafkaTests.swift

+        let _ = Task {
+            try await producer.run()
+        }


Let's use task groups here instead of unstructured tasks

FranzBusch · 2023-06-08T13:20:25Z

Tests/SwiftKafkaTests/KafkaPollingSystemTests.swift

+        self.runTask = Task {
+            try await self.sut.run(pollInterval: self.pollInterval)
+        }


FranzBusch · 2023-06-08T13:20:57Z

Tests/SwiftKafkaTests/KafkaPollingSystemTests.swift

+            XCTFail()
+        }
+
+        self.sut.produceMore()


Let's not call these directly but rather consume the sequence to trigger produce more

Motivation: We don't want to expose source and poll closure from `KafkaPollingSystem`. Modifications: * add new `func initialize(backPressureStrategy:, pollClosure:)` to `KafkaPollingSystem` to support delayed initialization (needed because the `AsyncSequence` relies on `KafkaPollingSystem` and `KafkaPollingSystem` relies on the `AsyncSequence.Source` (circular dependency) * `KafkaPollingSystem.StateMachine`: add state `.uninitialized` for when the `StateMachine` exists but `initialize()` has not yet been invoked on it * stop exposing `pollClosure` and `source` variables from `KafkaPollingSystem` / `KafkaPollingSystem.StateMachine` * move `yield(_:)` state functionality inside of `KafkaPollingSystem.StateMachine` * make `RDKafkaConfig.convertMessageToAcknowledgementResult` `private` * add new error type: `KafkaError.pollLoop` * throw error when invoking `KafkaPollingSystem.run()` on closed poll loop

Modifications: * refactor all tests into using task groups (structured concurrency) * adjust `KafkaPollingSystemTests` to new interface * add comments to task groups in `README` * Bug Fix: we were hotlooping because of missing return statements in `KafkaPollingSystem`'s poll loop

Motivation: When testing with `swift test` in terminal, the test `testRunTaskCancellationThrowsRunError` created a leaking continuation. This was due to the fact that the run task was still creating a new continuation although it has been cancelled. Modifications: * add a `guard` statement in `KafkaPollingSystem.run()` to check if `Task` is cancelled before creating a new `continuation`

felixschlegel · 2023-06-09T13:44:51Z

Tests/SwiftKafkaTests/KafkaProducerTests.swift

+
+                // This subscribes to the acknowledgements stream and immediately terminates the stream.
+                // Required to kill the run task.
+                var iterator: KafkaAsyncSequence<KafkaProducer.Acknowledgement>.AsyncIterator? = producer


I'm not extremely happy with this, but at the moment we need to subscribe to and then terminate the AsyncSequence to shut down the run task.

This kinda feels bad — do you have any ideas on how we can stop the run task from running forever when the user consumes no acknowledgement messages? @FranzBusch

Since we dropping the iterator here the didTerminate callback on the delegate should be triggered which should lead to the run() task to terminate. Can we check that the didTerminate is triggered?

The solution I came up with now is quite simple: I am just invoking self.pollingSystem.terminate() inside of KafkaProducer.shutDownGracefully(). By doing that we don't have to worry about whether the user subscribes to our acknowledgements AsyncSequence or not.

Motivation: By creating the `KafkaProducer` with a factory method that also returns the `KafkaAsyncSequence` for receiving message acknowledgements, we ensure that `KafkaAsyncSequence` will eventually get terminated as the user now owns the memory reference to it. It is important that the `KafkaAsyncSequence` gets terminated because otherwise the `KafkaProducer`'s run task runs forever. Modifications: * make initializer of `KafkaProducer` `private` * create new factory mehtod `KafkaProducer.newProducer` for initializing the `KafkaProducer` alongside the acknowledgement `KafkaAsyncSequence` * update tests * update README

This reverts commit 9880f63.

Motivation: Previously, we had to wait for the `KafkaProducer.acknowledgements` `AsyncSequence` delegate method `didTerminate()` to be invoked in order to shutdown the poll loop. This was an issue as this lead to the poll loop never stopping when e.g. the user never subscribed to the `acknowledgements` `AsyncSequence` and therefore the `didTerminate()` delegate method would never be invoked. Modifications: * make `KafkaPollingSystem.terminate()` `internal` * invoke `pollingSystem.terminate()` in `KafkaProducer.shutdownGracefully` * remove poll loop shutdown workaround in test Result: After your change, what will change.

FranzBusch · 2023-06-12T13:45:54Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+import NIOCore
+
+/// Our `AsyncSequence` implementation wrapping an `NIOAsyncSequenceProducer`.
+public struct KafkaAsyncSequence<Element>: AsyncSequence {


Just noticed that this is public. I would rather have two concrete AsyncSequences than a generic one here. We can use the generic one internally though. For this PR that would mean a KafkaMessageAcknowledgements struct

FranzBusch · 2023-06-12T15:16:07Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+    }
+
+    /// Kill the ``KafkaPollingSystem``'s poll loop and free its resources.
+    func terminate(_ error: Error? = nil) {


Can this be private?

Yes this would be no problem because we can also invoke didTerminate() from outside, but I thought that the name terminate() made more sense, also because didTerminate() is a delegate method.

(My implementation needs an internal method for terminating the KafkaPollingSystem because I invoke this from KafkaPollingSystem.shutDownGracefully)

FranzBusch · 2023-06-12T15:16:19Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+    /// Determines the next action to be taken in the poll loop based on the current state.
+    ///
+    /// - Returns: The next action for the poll loop.
+    func nextPollLoopAction() -> KafkaPollingSystem.StateMachine.PollLoopAction {


Do we really need a separate method for this?

Not really, but given that our stateMachine is private this provides a safe way to access the poll loop's state for testing. There are not too many things we can assert on here in testing. Alternatively we can make KafkaPollingSystem.yield(_:) return the next action but this would also then just serve the purpose on being able to assert on the KafkaPollingSystem. In general all of this is quite hard as the KafkaPollingSystem does not expose that much information to the outside world.

FranzBusch · 2023-06-12T15:18:01Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+            case alreadyClosed
+            /// The poll loop is already running. ``run()`` should not be invoked more than once!
+            case alreadyRunning
+            /// The poll loop can now be started.
+            case startLoop


One last nit here w.r.t. naming. Can you make sure all actions are named in way that they tell what do to e.g. throwError like you named startLoop

FranzBusch · 2023-06-12T15:18:59Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+                    return .alreadyRunning
+                }
+                self.state = .idle(source: source, pollClosure: pollClosure, running: true)
+            case .producing(let source, let pollClosure, let running):


Can this happen? How can we be in producing before run was called?

I thought this might be possible e.g. when produceMore() gets called upon the NIOAsyncSequenceProducerDelegate before run() gets invoked

That won't happen. produceMore() only gets called after yield() returns stopProducing

FranzBusch · 2023-06-12T15:19:36Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+        /// The possible actions for the poll loop.
+        enum PollLoopAction {
+            /// Ask `librdkakfa` to receive new message acknowledgements at a given poll interval.
+            case pollAndSleep(pollClosure: (() -> Void)?)


This can't be optional right?

FranzBusch · 2023-06-12T15:20:03Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+                source: Producer.Source?,
+                pollClosure: (() -> Void)?,


Why are the source and the pollClosure optional here?

You're right, this was from a previous implementation!

FranzBusch · 2023-06-12T15:21:09Z

Sources/SwiftKafka/KafkaPollingSystem.swift

+            case .idle(let source, _, _), .producing(let source, _, _), .stopProducing(let source, _, _, _):
+                // We can also yield when in .stopProducing,
+                // the AsyncSequenceProducer will buffer for us
+                let yieldResult = source?.yield(element)


Please don't yield inside the state machine. Yielding is a side effect so please return that as an action.

FranzBusch · 2023-06-12T15:21:52Z

Sources/SwiftKafka/KafkaProducer.swift

 /// Send messages to the Kafka cluster.
 /// Please make sure to explicitly call ``shutdownGracefully(timeout:)`` when the ``KafkaProducer`` is not used anymore.
 /// - Note: When messages get published to a non-existent topic, a new topic is created using the ``KafkaTopicConfig``
 /// configuration object (only works if server has `auto.create.topics.enable` property set).
 public actor KafkaProducer {
+    public typealias Acknowledgement = Result<KafkaAcknowledgedMessage, KafkaAcknowledgedMessageError>


Why are we typealiasing this here? Let's not introduce this please but spell it out

FranzBusch · 2023-06-12T15:29:51Z

Sources/SwiftKafka/KafkaProducer.swift

+        // Kill poll loop in polling system
+        self.pollingSystem.terminate()


I am not sure that this is enough. We should figure out what happens in a system where we don't poll anymore but still produce e.g. nobody consumes the acknowledgements sequence and it gets back pressured. Will rdkafka buffer then and at some point also back pressure or will we just OOM?

I believe rd_kafka_flush() got us covered here (we already use this).

Modifications: * `KafkaPollingSystem`: add `return` statement to every individual `switch` case * remove unneccesary optional types * remove typealias `KafkaProducer.Acknowledgement` * move `StateMachine.yield` side effect outside of `StateMachine` * rename `StateMachine.RunAction`s to be more imperative

Modifications: * remove generic `KafkaAsyncSequence` and have a specialised `AsyncSequence` called `KafkaMessageAcknowledgements` for the `KafkaProducer`

Motivation: Yielding before running resulted in a fatalError. Modifications: * remove message yields from testRunTaskCancellationThrowsRunError

This reverts commit 95c8dda.

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

felixschlegel · 2023-06-23T13:41:24Z

Having back pressure makes no sense as librdkafka is queuing the messages regardless, closing, but keeping branch for reference

FranzBusch reviewed May 30, 2023

View reviewed changes

felixschlegel commented May 31, 2023

View reviewed changes

felixschlegel requested a review from FranzBusch May 31, 2023 13:56

FranzBusch requested changes Jun 1, 2023

View reviewed changes

felixschlegel added 11 commits June 1, 2023 12:54

SwiftFormat

dcd7e68

* StateMachine: consolidated states .suspended and .loopSuspended int…

b5d8868

…o one

SwiftFormat

fb32e8b

BackPressurePolling: Inject polling closure

be38581

Motivation: We want to be able to inject a closure that contains the polling mechanism into our `KafkaBackPressurePollingSystem` so that we can assert that it has been called in testing.

Bug Fix: KafkaBackPressurePollingSystem

35e0da1

Condition for checking leaking continuations in `KafkaBackPressurePollingSystem.StateMachine.suspendLoop` was wrong.

SwiftFormat

f434b91

testNoPollsAfterPollLoopSuspension

52f572e

Modifications: * remove flakey tests testing that poll is not invoked when loop is suspended * add dedicated test that tests the case mentioned above

Remove TODOs

b7d5fec

felixschlegel force-pushed the fs-kafka-producer-poll-refactoring branch from ca5bc3d to b7d5fec Compare June 1, 2023 11:56

felixschlegel added 10 commits June 1, 2023 14:58

* KafkaBackPressurePollingSystem -> `KafkaAcknowledgementPollingSys…

77bbfba

…tem`

* remove TODOs

8487f40

* make KafkaAcknowledgemnetPollingSystem @unchecked Sendable

7196c07

* SwiftFormat

68c50eb

* add comments

625151a

felixschlegel requested a review from FranzBusch June 2, 2023 13:55

felixschlegel force-pushed the fs-kafka-producer-poll-refactoring branch from 521d9b2 to 52e8d25 Compare June 7, 2023 14:44

felixschlegel requested a review from FranzBusch June 7, 2023 14:46

* remove XCTWaiter (busted on CI)

486cfb2

felixschlegel force-pushed the fs-kafka-producer-poll-refactoring branch from 5da5e10 to 486cfb2 Compare June 7, 2023 15:29

felixschlegel added 3 commits June 7, 2023 16:38

* fix typos

d9f5abf

Make KafkaPollingSystem.pollClosure settable

80776fd

Modifications: * remove `ClosureWrapper` classes * make `pollClosure` `Optional` * make `pollClosure` a get/set property of `KafkaPollingSystem` * adjust tests

Refactor KafkaPollingSystemTests

797e35a

Modifications: * move common logic into `setUp()` method

FranzBusch reviewed Jun 8, 2023

View reviewed changes

felixschlegel force-pushed the fs-kafka-producer-poll-refactoring branch from d73c035 to 30fe1bf Compare June 8, 2023 16:35

felixschlegel added 3 commits June 9, 2023 14:37

Remove TODO

1e0c803

felixschlegel commented Jun 9, 2023

View reviewed changes

felixschlegel requested a review from FranzBusch June 9, 2023 13:45

felixschlegel added 4 commits June 9, 2023 16:25

Revert "Create KafkaProducer with factory method"

95c8dda

This reverts commit 9880f63.

* make client property of KafkaProducer immutable

7a16d21

felixschlegel force-pushed the fs-kafka-producer-poll-refactoring branch from dfbbe7e to 5e295df Compare June 12, 2023 09:03

FranzBusch reviewed Jun 12, 2023

View reviewed changes

felixschlegel added 2 commits June 12, 2023 18:12

Remove generic KafkaAsyncSequence

c4aca20

Modifications: * remove generic `KafkaAsyncSequence` and have a specialised `AsyncSequence` called `KafkaMessageAcknowledgements` for the `KafkaProducer`

felixschlegel requested a review from FranzBusch June 12, 2023 17:25

felixschlegel added 3 commits June 12, 2023 18:37

Fix testRunTaskCancellationThrowsRunError

cc0e14c

Motivation: Yielding before running resulted in a fatalError. Modifications: * remove message yields from testRunTaskCancellationThrowsRunError

Revert "Revert "Create KafkaProducer with factory method""

9c34dab

This reverts commit 95c8dda.

felixschlegel force-pushed the fs-kafka-producer-poll-refactoring branch from de0d8e3 to ff2cb71 Compare June 14, 2023 13:03

felixschlegel closed this Jun 23, 2023

		private let _client: NIOLockedValueBox<KafkaClient?>
		private let _sequenceSource: NIOLockedValueBox<Producer.Source?>

	fatalError("Created leaking continuation")
	fatalError("Internal state inconsistency. Run loop is running more than once")

	static func convertMessageToAcknowledgementResult(
	private static func convertMessageToAcknowledgementResult(

		// Kill poll loop in polling system
		self.pollingSystem.terminate()

KafkaProducer: Backpressure for Ack'd messages #53

KafkaProducer: Backpressure for Ack'd messages #53

Conversation

felixschlegel commented May 26, 2023 • edited Loading

Motivation

Modifications

TODO

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

felixschlegel commented May 31, 2023

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

FranzBusch left a 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

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

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

felixschlegel Jun 12, 2023 • edited Loading

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

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

felixschlegel commented Jun 23, 2023

felixschlegel commented May 26, 2023 •

edited

Loading

felixschlegel Jun 12, 2023 •

edited

Loading