Implementation of the `KafkaConsumer` #24

felixschlegel · 2022-09-06T14:01:52Z

Modifications:

implemented the KafkaConsumer

KafkaConsumer has two public initializers:

// Subscribe to topics as part of a consumer group
KafkaConsumer(
    topics: [String],
    groupID: String,
    config: KafkaConfig = KafkaConfig(),
    logger: Logger
) async throws {}

// Assign consumer to a particular topic partition
KafkaConsumer(
    topic: String,
    partition: KafkaPartition,
    offset: Int64 = Int64(RD_KAFKA_OFFSET_END),
    config: KafkaConfig = KafkaConfig(),
    logger: Logger
) async throws {

incoming messages can be read using KafkaConsumer's messages AsyncSequence that has backpressure implemented
added the ability for manual offset management by implementing the commitSync() method for KafkaConsumer
implemented Integration Tests

Result:

Users can consume messages from a Kafka cluster.

Sources/SwiftKafka/KafkaConsumer.swift

felixschlegel · 2022-09-11T17:53:18Z

Sources/SwiftKafka/KafkaConsumer.swift

+    func poll(timeout: Int32 = 100) async throws -> KafkaConsumerMessage? {
+        // TODO: clock API
+        // TODO: ideally: make private
+        // TODO: is this state needed here?


This is a good candidate for Clock API — shall I create a ticket?

Again, this function is not meant to be used from outside the KafkaConsumer — we could therefore think about omitting state here as well

I don't think we need the Clock API here. There is no real value as far as I can see

Yeah let's make it private and remove the state

felixschlegel · 2022-09-11T17:54:30Z

Sources/SwiftKafka/KafkaConsumer.swift

+        try await self._commitSync(message)
+    }
+
+    // TODO: commit multiple messages at once -> different topics + test


What do you think about refactoring this function into a function that takes an array of KafkaConsumerMessage as a parameter and commits all of them at once?

This would then always commit the highest offset for each topic.

felixschlegel · 2022-09-11T17:55:00Z

Sources/SwiftKafka/KafkaConsumerMessage.swift

+//            let errorString = errorStringBuffer.readString(length: errorStringBuffer.readableBytes)
+
+            // TODO: what to do with error string?
+            // TODO: handle errors here or in consumer?


Handle errors here or in KafkaConsumer?

I think handling it here and throwing is fine, but shouldn't we do it as the first thing in this method?

We could move it up but we would have to move up the valuePointer and the valueBufferPointer as well — I'll implement it in my next commit

Tests/IntegrationTests/SwiftKafkaTests.swift

Sources/SwiftKafka/KafkaConsumer.swift

Modifications: * created a new .testTarget "IntegrationTests" with a dedicated directory * made KafkaConsumerMessage and KafkaAcknowledgedMessage use KafkaPartition as type for the partition property * implemented first consumer that enables users to subscribe to topics and poll the consumer * built a KafkaSequenceConsumer on top of KafkaConsumer that enables users to consume messages in an AsyncSequence * enforced that the integration-test-topic is created when in the Kafka docker container Result: Users are able to consume messages from a Kafka cluster

Modifications: * created Tests for KafkaConsumer * updated Docker compose Kafka server to create different topics for different tests to avoid overlaps * introduced two initializers for KafkaConsumer allowing to subscribe to a topic as a group or assign the KafkaConsumer to a particular topic+partition pair * added ability to manually commit offsets to Kafka

Modifications: * made KafkaConsumer an actor * implemented backpressure for KafkaConsumer

Modifications: * made KafkaConsumer.close() method async instead of blocking

mr-swifter · 2022-09-16T16:44:12Z

Hello @felixschlegel !

I was looking at the PR have one small question, just curious.

Do you have plans to address that now some lifrdkafka's blocking functions are called from Tasks? I see, in particular:

rd_kafka_commit from commitSync/_commitSync
rd_kafka_consumer_poll from produceMore

Or it's beyond the scope? That would be very nice to make the client truly behave async.

I would be happy to discuss how that can be addressed.

felixschlegel · 2022-09-19T10:40:27Z

Hello @felixschlegel !

I was looking at the PR have one small question, just curious.

Do you have plans to address that now some lifrdkafka's blocking functions are called from Tasks? I see, in particular:

rd_kafka_commit from commitSync/_commitSync

rd_kafka_consumer_poll from produceMore

Or it's beyond the scope? That would be very nice to make the client truly behave async.

I would be happy to discuss how that can be addressed.

Hey @mr-swifter,

Nice to hear that you are getting involved in developing this package!

Thanks for pointing out that the Tasks might get us into trouble. In this comment, I already addressed the desire to get rid of the Task for the produceMore method.

It is possible to make the commitSync method behave async in a non-blocking manner. However, we would only be able to know if the commit was successfully scheduled and not if the commit itself was successful.

I hope I understood your comment correctly. Please let me know what you think!

mr-swifter · 2022-09-19T15:53:17Z

Hi @felixschlegel !

Yep, it seems I missed this comment when was looking at changes. It's good if you have some plans to fix this!

Regarding commitSync I mean that it would be nice to make the call behave that await consumer.commitSync will be returned back only when commit is send and has been replied by kafka, so client can return proper result to user, but meantime the Task where commitSync happens will be properly suspended, but not sitting on blocking call, so the thread can execute other tasks meanwhile.

I do something like this at my version, maybe not perfect, but works in my case and provide desired behaviour:

func commitSync(...) async { 
    await withCheckedContinuation { continuation in
        kafkaDispatchQueue.async { // shedule executution to separate queue managed by my kafka client package outside of cooperative pool 
            let result = rd_kafka_commit(...) // or any other blocking or timeout-based blocking call like rd_kafka_consumer_poll
            /// wrap-up the result if needed 
            continuation.resume(returning: result)
        }
    }
}

...
let result = await consumer.commitSync()
/// here we know the result of commit

So, basically I use separate execution queue for scheduling blocking calls and use continuation to resume task execution when librdkafka blocking function is over.
At the moment it's single queue, but I see maybe I will need to migrate to a few queue for different purposes (like poll vs commits vs transactions etc).

FranzBusch · 2022-09-20T08:33:18Z

@mr-swifter Thanks for raising this. I fully agree with you that we should not block any of the cooperative thread pools threads here since it can lead to various problems and the proper thing to do is actually back the consumer by a queue and dispatch every call onto this queue. So that we only block the thread backing the queue and just take a continuation that will be resumed once the blocking call finished.

The interesting challenge here is going to be handling cancellation; however, in the beginning we could just punt that and implement that at a later point. Having a queue is already a bigger refactor and brings us a lot of benefits.

Modifications: * KafkaConsumer blocking operations are now executed on a serial `DispatchQueue` * remove `deinit` as `close` is a blocking function and shall be invoked by the user anyway

felixschlegel · 2022-09-20T13:19:28Z

Hey @mr-swifter , Hey @FranzBusch ,

thank you for your comments! I have implemented a single serial DispatchQueue that takes care of all our blocking methods. Let me know what you think!

FranzBusch

Looks really good already, I left a few comments inline. There are a few larger points which I would like us to track in issues.

We should think again if we may want to split this type into KafkaSinglePartitionConsumer and KafkaGroupConsumer
librdkafka exposes a callback based API for consuming messages which ought to be faster than the poll based one. We should investigate this

FranzBusch · 2022-09-12T12:52:06Z

Sources/SwiftKafka/KafkaConsumerMessage.swift

+//            let errorString = errorStringBuffer.readString(length: errorStringBuffer.readableBytes)
+
+            // TODO: what to do with error string?
+            // TODO: handle errors here or in consumer?


I think handling it here and throwing is fine, but shouldn't we do it as the first thing in this method?

Sources/SwiftKafka/KafkaConsumer.swift

FranzBusch · 2022-09-21T12:28:12Z

Sources/SwiftKafka/KafkaConsumer.swift

+    private typealias Element = Result<KafkaConsumerMessage, Error> // TODO: replace with a more specific Error type
+    private let messagesSource: NIOAsyncSequenceProducer<
+        Element,
+        ConsumerMessagesAsyncSequence.HighLowWatermark,
+        ConsumerMessagesAsyncSequenceDelegate
+    >.Source


Wondering if we should make this a throwing sequence instead of result based. Always hard to say what happens if a consumed message produces an error if more messages can come afterward

Can we postpone this topic?

FranzBusch · 2022-09-21T13:21:34Z

Sources/SwiftKafka/KafkaConsumer.swift

+    func poll(timeout: Int32 = 100) async throws -> KafkaConsumerMessage? {
+        // TODO: clock API
+        // TODO: ideally: make private
+        // TODO: is this state needed here?


I don't think we need the Clock API here. There is no real value as far as I can see

Yeah let's make it private and remove the state

FranzBusch · 2022-09-21T13:22:32Z

Sources/SwiftKafka/KafkaConsumer.swift

+    /// This method blocks for a maximum of `timeout` milliseconds.
+    /// - Parameter timeout: Maximum amount of milliseconds this method waits for a new message.
+    /// - Returns: A ``KafkaConsumerMessage`` or `nil` if there are no new messages.
+    func poll(timeout: Int32 = 100) throws -> KafkaConsumerMessage? {


Can we add a precondition in this method to check that we are on the current serial queue please

FranzBusch · 2022-09-21T13:23:27Z

Sources/SwiftKafka/KafkaConsumer.swift

+        try await withCheckedThrowingContinuation { continuation in
+            self.serialQueue.async {
+                body(continuation)
+            }
+        }


We should add a note here that we are not supporting cancellation at the moment and that we need to fix this up. Please also create an issue for this.

FranzBusch · 2022-09-21T13:24:06Z

Sources/SwiftKafka/KafkaConsumer.swift

+    }
+
+    /// Stop consuming messages. This step is irreversible.
+    public func close() async throws {


Should we really expose this? IMO we should just do this through the didTerminate callback on the AsyncSequence.

FranzBusch · 2022-09-21T13:25:57Z

Sources/SwiftKafka/KafkaConsumer.swift

+    // TODO: commit multiple messages at once -> different topics + test
+    // TODO: docc: https://github.com/segmentio/kafka-go#explicit-commits note about highest offset
+    private func _commitSync(_ message: KafkaConsumerMessage) throws {
+        guard self.config.value(forKey: "enable.auto.commit") == "false" else {


Can we add a precondition for the queue here as well please

mr-swifter · 2022-09-22T10:42:40Z

Great stuff! Thanks @felixschlegel , @FranzBusch !

Modifications: * move from Swift 5.6 to Swift 5.7 * add offset property to assignment consumer initializer * remove default config object in consumer initializer * added preconditions in KafkaConsumer that check if functions are called on the serial queue * make ConsumerMessagesAsyncSequenceDelegate a struct containing closures * close KafkaConsumer on NIOAsyncSequenceProducerDelegate.didTerminate instead of a public close function * add test utility functions to create and delete topics * Docker: build librdkafka from source instead of installing from apt because apt only provides librdkafka version 1.2.1 whereas we use 1.9.2 currently * repalce KafkaConsumer.State with simple boolean "closed" and assertions * refactor KafkaConsumerMessage * test IntegrationTests with multiple messages for each test

felixschlegel · 2022-09-23T13:22:42Z

Hey @FranzBusch,

I have implemented your suggested changes and created new issues for some of your comments. Please let me know what you think!

With the new topic creation method, I get this error when testing commitSync in the IntegrationTests:

%5|1663939258.150|PARTCNT|rdkafka#consumer-9| [thrd:main]: Topic 6B1E56D4-3387-4CEA-AC41-FB92471FBAFD partition count changed from 1 to 0
%4|1663939258.189|COMMITFAIL|rdkafka#consumer-9| [thrd:main]: Offset commit (unassigned partitions) failed for 1/1 partition(s) in join-state init: Broker: Unknown topic or partition: 6B1E56D4-3387-4CEA-AC41-FB92471FBAFD[0]@10(Broker: Unknown topic or partition)

However, the test still succeeds. Just wanted to let you know, maybe you have an idea what's going on here!

felixschlegel · 2022-09-23T21:06:10Z

Sources/SwiftKafka/KafkaConsumer.swift

+            let messageResult: Element
+            do {
+                guard let message = try self.poll() else {
+                    self.produceMore()


@FranzBusch
Is this behaviour correct or will the NIOAsyncSequence invoke produceMore() again anyway?

I think for the shape that we are doing here it is okay to call self.produceMore() again. The NIOAsyncSequenceProducer will only call produceMore once when an element has been consumed and you need to make sure that at some point that leads to a new message.

FranzBusch

Let's get this merged. I think we made some good progress here on an initial consumer and opened a few issues to improve the API.

FranzBusch · 2022-09-26T12:56:48Z

Sources/SwiftKafka/KafkaConsumer.swift

+        let messagesSequenceDelegate = ConsumerMessagesAsyncSequenceDelegate { [weak self] in
+            self?.produceMore()
+        } didTerminateClosure: { [weak self] in
+            self?.close()
+        }


We need to find a way around weak here can we open an issue for this please. weak has some serious performance implications sadly.

FranzBusch · 2022-09-26T13:04:41Z

Sources/SwiftKafka/KafkaConsumer.swift

+            let messageResult: Element
+            do {
+                guard let message = try self.poll() else {
+                    self.produceMore()


I think for the shape that we are doing here it is okay to call self.produceMore() again. The NIOAsyncSequenceProducer will only call produceMore once when an element has been consumed and you need to make sure that at some point that leads to a new message.

felixschlegel force-pushed the fs-kafka-consumer branch 5 times, most recently from 88b2742 to 8f0a618 Compare September 11, 2022 17:34

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Tests/IntegrationTests/SwiftKafkaTests.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel commented Sep 11, 2022

View reviewed changes

Sources/SwiftKafka/KafkaConsumer.swift Outdated Show resolved Hide resolved

felixschlegel marked this pull request as ready for review September 11, 2022 18:01

felixschlegel changed the title ~~Implementation of the Kafka Consumers~~ Implementation of the KafkaConsumer Sep 11, 2022

felixschlegel mentioned this pull request Sep 12, 2022

KafkaProducer with sendAsync method #21

Merged

felixschlegel added 4 commits September 16, 2022 13:34

Implemented Backpressure into KafkaConsumer

f906fe2

Modifications: * made KafkaConsumer an actor * implemented backpressure for KafkaConsumer

DocC Comments for KafkaConsumer

e5bf998

Modifications: * made KafkaConsumer.close() method async instead of blocking

felixschlegel force-pushed the fs-kafka-consumer branch from 8f0a618 to e5bf998 Compare September 16, 2022 11:35

Improve Concurrency Model for KafkaConsumer

1133b59

Modifications: * KafkaConsumer blocking operations are now executed on a serial `DispatchQueue` * remove `deinit` as `close` is a blocking function and shall be invoked by the user anyway

FranzBusch reviewed Sep 21, 2022

View reviewed changes

felixschlegel requested a review from FranzBusch September 23, 2022 13:29

felixschlegel commented Sep 23, 2022

View reviewed changes

FranzBusch approved these changes Sep 26, 2022

View reviewed changes

FranzBusch merged commit ae5982c into swift-server:main Sep 26, 2022

felixschlegel mentioned this pull request Sep 26, 2022

Replace weak reference to KafkaConsumer in ConsumerMessagesAsyncSequenceDelegate #37

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implementation of the `KafkaConsumer` #24

Implementation of the `KafkaConsumer` #24

felixschlegel commented Sep 6, 2022 •

edited

Loading

felixschlegel Sep 11, 2022

FranzBusch Sep 21, 2022

felixschlegel Sep 11, 2022

felixschlegel Sep 11, 2022

FranzBusch Sep 12, 2022

felixschlegel Sep 23, 2022

mr-swifter commented Sep 16, 2022

felixschlegel commented Sep 19, 2022 •

edited

Loading

mr-swifter commented Sep 19, 2022

FranzBusch commented Sep 20, 2022

felixschlegel commented Sep 20, 2022

FranzBusch left a comment

FranzBusch Sep 12, 2022

FranzBusch Sep 21, 2022

felixschlegel Sep 23, 2022

FranzBusch Sep 21, 2022

FranzBusch Sep 21, 2022

FranzBusch Sep 21, 2022

felixschlegel Sep 23, 2022

FranzBusch Sep 21, 2022

FranzBusch Sep 21, 2022

mr-swifter commented Sep 22, 2022

felixschlegel commented Sep 23, 2022

felixschlegel Sep 23, 2022

FranzBusch Sep 26, 2022

FranzBusch left a comment

FranzBusch Sep 26, 2022

felixschlegel Sep 26, 2022

FranzBusch Sep 26, 2022

Implementation of the KafkaConsumer #24

Implementation of the KafkaConsumer #24

Conversation

felixschlegel commented Sep 6, 2022 • edited Loading

Modifications:

Result:

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

mr-swifter commented Sep 16, 2022

felixschlegel commented Sep 19, 2022 • edited Loading

mr-swifter commented Sep 19, 2022

FranzBusch commented Sep 20, 2022

felixschlegel commented Sep 20, 2022

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

mr-swifter commented Sep 22, 2022

felixschlegel commented Sep 23, 2022

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

Implementation of the `KafkaConsumer` #24

Implementation of the `KafkaConsumer` #24

felixschlegel commented Sep 6, 2022 •

edited

Loading

felixschlegel commented Sep 19, 2022 •

edited

Loading