`KafkaConsumer` Refactoring #66

felixschlegel · 2023-06-19T16:43:52Z

Motivation

Our KafkaConsumer should expose a run() method that serves the consumer queue and polls librdkafka for new messages and any other queued callbacks. This aligns with the new implementation of KafkaProducer.

Additionally

get rid of weak reference to self in KafkaConsumer
get rid of DispatchQueue in KafkaConsumer

Modifications

expose new method KafkaConsumer.run()
write async wrapper method KafkaClient.consumerPoll(timeout:)
replace serialQueue in KafkaConsumer with StateMachine that
encapsulates all variables and can also be accessed from the
ConsumerMessagesAsyncSequence
close KafkaConsumer when for await loop of AsyncSequence is
exited
make KafkaConsumer.shutdownGracefully public
remove BackPressureStrategy from KafkaConsumerConfiguration

FranzBusch · 2023-06-26T10:29:47Z

Sources/SwiftKafka/KafkaClient.swift

+    /// - Parameter timeout: Maximum amount of milliseconds this method waits for a new message.
+    /// - 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(timeout: Int32) async throws -> KafkaConsumerMessage? {


We should remove the timeout here otherwise people might actually block. Additionally, we should remove the async from this method. I don't see a reason why we are creating a continuation here. We should be able to just call rd_kafka_consumer_poll and bridge the message into a Swift type, destroy the pointer and return.

Totally makes sense!

Modifications: * `KafkaConsumer`: remove `NIOAsyncSequenceProducer` and use own implementation of `AsyncSequence` instead * write async wrapper method `KafkaClient.consumerPoll(timeout:)` * update memory leak tests for `KafkaConsumer` and `KafkaProducer` * update tests

Modifications: * replace `serialQueue` in `KafkaConsumer` with `StateMachine` that encapsulates all variables and can also be accessed from the `ConsumerMessagesAsyncSequence` * close `KafkaConsumer` when `for await` loop of `AsyncSequence` is exited * make `ConsumerMessagesAsyncIterator` a class-backed `struct` * make `KafkaConsumer.shutdownGracefully` `public`

Modifications: * add `run()` method to `KafkaConsumer` * rename `ConsumerMessagesAsyncSequence` to `KafkaConsumerMessages` * `KafkaConsumer`: put back `NIOAsyncSequenceProducer` * update `README` * use `Duration` type for `pollInterval`

Modifications: * `KafkaClient.consumerPoll`: remove `timeout` as this method should only be used with `timeout = 0` * `KafkaClient.consumerPoll`: remove continuation as it has no benefit

FranzBusch · 2023-06-26T10:39:40Z

Sources/SwiftKafka/KafkaConsumer.swift

+    ///
+    /// - Note: Invoking this function is not always needed as the ``KafkaConsumer``
+    /// will already shut down when consumption of the ``KafkaConsumerMessages`` has ended.
+    public func shutdownGracefully() {


We should make this method private and since we intend to use service lifecycle to do this in the run() method.

FranzBusch · 2023-06-26T13:14:33Z

Sources/SwiftKafka/KafkaConsumer.swift

+            let result = client.withKafkaHandlePointer { handle in
+                rd_kafka_subscribe(handle, subscribedTopicsPointer)
+            }


For all these places where we use withKafkaHandlePointer can we just add a method on the KafkaClient instead? At best that method also already does the conversion to the correct error.

FranzBusch · 2023-06-26T13:23:52Z

Sources/SwiftKafka/KafkaConsumer.swift

+                    }
+                    messageResult = .success(message)
+                } catch let kafkaError as KafkaError {
+                    messageResult = .failure(kafkaError)


If we fail polling shouldn't we shutdown everything?

So consumerPoll throws when there is an error with the received message. However, I think we shouldn't close the entire consumer because of, e.g. a message with a faulty payload.

Hm, I think this is pretty bad. If we get a bad message something seriously went wrong. At least that is what I expect. Can you investigate when this could happen and document it here. IMO it is weird that our sequence contains a Result<>, I would like to get rid of that.

FranzBusch · 2023-06-26T13:24:10Z

Sources/SwiftKafka/KafkaConsumer.swift

            do {
-                try self._commitSync(message)
+                try self._commitSync(message) // Blocks until commiting the offset is done


Yeah that is problematic. We need to change that.

FranzBusch · 2023-06-26T13:25:19Z

Sources/SwiftKafka/KafkaConsumer.swift

+                client: KafkaClient,
+                source: Producer.Source,
+                subscribedTopicsPointer: UnsafeMutablePointer<rd_kafka_topic_partition_list_t>,
+                logger: Logger


Let's not store the logger in here. We can just keep it in a separate prop since it is not state.

FranzBusch · 2023-06-26T13:25:37Z

Sources/SwiftKafka/KafkaConsumer.swift

+                pollInterval: Duration,
+                client: KafkaClient,
+                source: Producer.Source,
+                subscribedTopicsPointer: UnsafeMutablePointer<rd_kafka_topic_partition_list_t>,


We should create an abstraction for this so we don't hold UnsafeMutablePointers all the time.

FranzBusch · 2023-06-26T13:25:54Z

Sources/SwiftKafka/KafkaConsumer.swift

+            /// - Parameter subscribedTopicsPointer: Pointer to a list of topics + partition pairs.
+            /// - Parameter logger: A logger.
+            case initializing(
+                pollInterval: Duration,


Doesn't need to be in the state machine since it is a fixed configuration.

FranzBusch · 2023-06-26T13:26:31Z

Sources/SwiftKafka/KafkaConsumer.swift

+            /// Kill the poll loop.
+            case killPollLoop


Kill is a bit harsh. Can we call this terminate instead?

Modifications: * make `KafkaConsumer.shutdownGracefully()` `private` * `KafkaConsumer`: create `librdkafka` wrapping methods in `KafkaClient` instead of using `KafkaClient.withKafkaHandlePointer` * `KafkaConsumer`: rename `.killPollLoop` -> `.terminatePollLoop` * `KafkaConsumer.StateMachine`: move `logger` out of `State` * `KafkaConsumer`: move `pollInterval` out of `StateMachine`

Modifications: * create new class `RDKafkaTopicPartitionList` wrapping a `rd_kafka_topic_partition_list_t`

FranzBusch · 2023-06-27T14:14:58Z

Sources/SwiftKafka/KafkaConsumer.swift

+        ShutdownOnTerminate
+    >
+    /// Time between two consecutive polls.
+    private var pollInterval: Duration


Why is this not part of the KafkaConsumerConfiguration?

Wanted to do that in a separate PR as this should also be done for KafkaProducer, but yeah I can do it in this one already

Separate PR is fine

Too late now 😅

FranzBusch · 2023-06-27T14:20:21Z

Sources/SwiftKafka/KafkaConsumer.swift

+                    }
+                    messageResult = .success(message)
+                } catch let kafkaError as KafkaError {
+                    messageResult = .failure(kafkaError)


Hm, I think this is pretty bad. If we get a bad message something seriously went wrong. At least that is what I expect. Can you investigate when this could happen and document it here. IMO it is weird that our sequence contains a Result<>, I would like to get rid of that.

FranzBusch · 2023-06-27T14:22:34Z

Sources/SwiftKafka/RDKafka/RDKafkaTopicPartitionList.swift

+import Crdkafka
+
+/// Swift wrapper type for `rd_kafka_topic_partition_list_t`.
+class RDKafkaTopicPartitionList {


Suggested change

class RDKafkaTopicPartitionList {

final class RDKafkaTopicPartitionList {

Modifications: * make `RDKafkaTopicPartitionList` a `final class` * make `KafkaConsumer.messages` `AsyncSequence` return `KafkaConsumerMessage` instead of `Result<,>` type * `KafkaConsumer` fails immediately when `rd_kafka_message_t` with error payload is received * move `KafkaConsumer.pollInterval` to `KafkaConsumerConfiguration` * move `KafkaProducer.pollInterval` to `KafkaProducerConfiguration` * update `README`

felixschlegel force-pushed the fs-kafka-consumer-poll-refactoring branch 2 times, most recently from b6aa400 to 0677dcf Compare June 21, 2023 13:41

felixschlegel mentioned this pull request Jun 22, 2023

Fix: KafkaConsumer.commitSync should not block #68

Closed

FranzBusch reviewed Jun 26, 2023

View reviewed changes

felixschlegel added 8 commits June 26, 2023 11:50

* remove KafkaConsumerTests

bc8a203

* remove unused NIOCore imports

a824f46

* remove backpressure strategy from KafkaConsumerConfig

3abb4d0

* KafkaConsumer: inject poll interval through initializer (temporary)

4796307

Move polling to dedicated run() method

8030c68

Modifications: * add `run()` method to `KafkaConsumer` * rename `ConsumerMessagesAsyncSequence` to `KafkaConsumerMessages` * `KafkaConsumer`: put back `NIOAsyncSequenceProducer` * update `README` * use `Duration` type for `pollInterval`

Review Franz

fa9ebf4

Modifications: * `KafkaClient.consumerPoll`: remove `timeout` as this method should only be used with `timeout = 0` * `KafkaClient.consumerPoll`: remove continuation as it has no benefit

felixschlegel force-pushed the fs-kafka-consumer-poll-refactoring branch from 0677dcf to fa9ebf4 Compare June 26, 2023 11:01

felixschlegel requested a review from FranzBusch June 26, 2023 11:02

FranzBusch reviewed Jun 26, 2023

View reviewed changes

felixschlegel added 2 commits June 26, 2023 17:12

Wrapper type for rd_kafka_topic_partition_list_t

4b2dce6

Modifications: * create new class `RDKafkaTopicPartitionList` wrapping a `rd_kafka_topic_partition_list_t`

felixschlegel requested a review from FranzBusch June 26, 2023 17:15

KafkaConsumer: Remove unneccesary references to subscribedTopicsPointer

a7ba78a

felixschlegel mentioned this pull request Jun 27, 2023

Functionality wishlist #75

Open

11 tasks

FranzBusch reviewed Jun 27, 2023

View reviewed changes

felixschlegel requested a review from FranzBusch June 27, 2023 16:20

FranzBusch approved these changes Jun 28, 2023

View reviewed changes

FranzBusch merged commit b0bc014 into swift-server:main Jun 28, 2023

FranzBusch deleted the fs-kafka-consumer-poll-refactoring branch June 28, 2023 17:15

blindspotbounty mentioned this pull request Sep 12, 2023

Enable back pressure for KafkaConsumer #131

Closed

	class RDKafkaTopicPartitionList {
	final class RDKafkaTopicPartitionList {

KafkaConsumer Refactoring #66

KafkaConsumer Refactoring #66

Uh oh!

Conversation

felixschlegel commented Jun 19, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Additionally

Modifications

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

felixschlegel Jun 27, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

`KafkaConsumer` Refactoring #66

`KafkaConsumer` Refactoring #66

felixschlegel commented Jun 19, 2023 •

edited

Loading

felixschlegel Jun 27, 2023 •

edited

Loading