`KafkaProducerMessage`: generic key & value #95

felixschlegel · 2023-07-21T16:31:35Z

Motivation:

We want to allow our users to feed arbitrary data as key and value
of a KafkaProducerMessage.

Modifications:

add new public protocol KafkaBuffer which asks implementing types to
provide safe access to an UnsafeMutableRawBufferPointer containing
the data
implement KafkaBuffer in String
implement KafkaBuffer in ByteBuffer
update tests

felixschlegel · 2023-07-21T16:33:14Z

Tests/SwiftKafkaTests/KafkaProducerTests.swift

@@ -110,7 +110,7 @@ final class KafkaProducerTests: XCTestCase {
            }

            let expectedTopic = "test-topic"
-            let message = KafkaProducerMessage(
+            let message = KafkaProducerMessage<ByteBuffer, ByteBuffer>(


Keys are optional, which leads to the problem that their type cannot be inferred when set to nil. This is a workaround but far from perfect. Any suggestions?

Can we conform Never to KafkaBuffer and provide an init that doesn't take a key when we are Never.

FranzBusch · 2023-07-24T15:31:10Z

Sources/SwiftKafka/Data/String+KafkaBuffer.swift

+    /// Yields a buffer pointer containing this `KafkaBuffer`'s readable bytes.
+    ///
+    /// - warning: Do not escape the pointer from the closure for later use.
+    ///
+    /// - parameters:
+    ///     - body: The closure that will accept the yielded bytes.
+    /// - returns: The value returned by `body`.


Just having this doc on the protocol should be enough.

FranzBusch · 2023-07-24T15:31:42Z

Sources/SwiftKafka/Data/String+KafkaBuffer.swift

+        try ByteBuffer(string: self).withUnsafeReadableBytes { UnsafeRawBufferPointer in
+            try body(UnsafeRawBufferPointer)
+        }


We should try to use self.utf8.withContiguousStorageIfAvailable and only fallback to bridging into ByteBuffer in the slow path

FranzBusch · 2023-07-24T15:31:57Z

Sources/SwiftKafka/Data/ByteBuffer+KafkaBuffer.swift

+    /// Yields a buffer pointer containing this `KafkaBuffer`'s readable bytes.
+    ///
+    /// - warning: Do not escape the pointer from the closure for later use.
+    ///
+    /// - parameters:
+    ///     - body: The closure that will accept the yielded bytes.
+    /// - returns: The value returned by `body`.


FranzBusch · 2023-07-24T15:32:15Z

Sources/SwiftKafka/KafkaProducer.swift

@@ -252,7 +252,7 @@ public final class KafkaProducer: Service, Sendable {
    /// of the corresponding ``KafkaAcknowledgedMessage``.
    /// - Throws: A ``KafkaError`` if sending the message failed.
    @discardableResult
-    public func send(_ message: KafkaProducerMessage) throws -> KafkaProducerMessageID {
+    public func send<K, V>(_ message: KafkaProducerMessage<K, V>) throws -> KafkaProducerMessageID {


Can we name those generic types Key and Value please

FranzBusch · 2023-07-24T15:32:19Z

Sources/SwiftKafka/RDKafka/RDKafkaClient.swift

@@ -97,33 +97,45 @@ final class RDKafkaClient: Sendable {
    /// - Parameter newMessageID: ID that was assigned to the `message`.
    /// - Parameter topicConfig: The ``KafkaTopicConfiguration`` used for newly created topics.
    /// - Parameter topicHandles: Topic handles that this client uses to produce new messages
-    func produce(
-        message: KafkaProducerMessage,
+    func produce<K, V>(


glbrntt · 2023-07-24T15:53:30Z

Sources/SwiftKafka/Data/KafkaBuffer.swift

+    /// - parameters:
+    ///     - body: The closure that will accept the yielded bytes.
+    /// - returns: The value returned by `body`.
+    func withUnsafeRawBufferPointer<T>(_ body: (UnsafeRawBufferPointer) throws -> T) rethrows -> T


Does this need to have such a low-level requirement? It feels like an API smell to me that the only requirement is an unsafe API.

Is it possible, for example, to rephrase the API in terms of e.g. RandomAccessCollection of UInt8 and have encoders/decoders?

I was thinking about this as well but I am unsure this fulfils the goals that we want to achieve here. We want to make it possible to have both key and especially value be able to pass copy-free to rdkafka. If we take a collection of UInt8 then I am not sure if we can achieve that.

You do have withContiguousStorageIfAvailable for RAC which should fulfil that most of the time (I guess only custom RACs might not implement it).

Yeah you are right we can probable be generic over RAC here instead of creating our own type. In the slow path we just write the RAC into byte buffer. @felixschlegel can you try this out.

I agree that having the extra KafkaBuffer type is not ideal. However, I do not find the RandomAccessCollection approach more ergonomic (from what I have investigated):

Our KafkaProducerMessage type's String initializer would have to take String as parameter value, but stores a ContiuousArray<UInt8> which still requires copying

See implementation

/// Message that is sent by the `KafkaProducer` public struct KafkaProducerMessage<Key: RandomAccessCollection<UInt8>, Value: RandomAccessCollection<UInt8>> { public var topic: String public var partition: KafkaPartition public var key: Key? public var value: Value /// Create a new `KafkaProducerMessage` with a key and value. /// - Parameter topic: The topic the message will be sent to. Topics may be created by the `KafkaProducer` if non-existent. /// - Parameter partition: The topic partition the message will be sent to. If not set explicitly, the partiotion will be assigned automatically. /// - Parameter key: Used to guarantee that messages with the same key will be sent to the same partittion so that their order is preserved. /// - Parameter value: The message body. public init( topic: String, partition: KafkaPartition? = nil, key: Key, value: Value ) { self.topic = topic self.key = key self.value = value if let partition = partition { self.partition = partition } else { self.partition = .unassigned } } } // MARK: - KafkaProducerMessage + String Initializer extension KafkaProducerMessage where Key == ContiguousArray<UInt8>, Value == ContiguousArray<UInt8> { /// Create a new `KafkaProducerMessage` with a key and value. /// - Parameter topic: The topic the message will be sent to. Topics may be created by the `KafkaProducer` if non-existent. /// - Parameter partition: The topic partition the message will be sent to. If not set explicitly, the partiotion will be assigned automatically. /// - Parameter key: Used to guarantee that messages with the same key will be sent to the same partittion so that their order is preserved. /// - Parameter value: The message body. public init( topic: String, partition: KafkaPartition? = nil, key: String? = nil, value: String ) { self.topic = topic self.key = key.map { ContiguousArray($0.utf8) } self.value = ContiguousArray(value.utf8) if let partition = partition { self.partition = partition } else { self.partition = .unassigned } } }

We would have to take ByteBufferView as parameter type instead of ByteBuffer as only ByteBufferView conforms to RandomAccessCollection

So yes it is a smell, but we provide default implementations of KafkaBuffer for String and ByteBuffer. The protocol gives the user the freedom to make their own types conform to KafkaBuffer at their own risk. Furthermore this enables them to avoid copying their own types.

Ah right, UTF8View is not a RAC, but it is a BidirectionalCollection which should be sufficient to avoid the copy. I don't think having convenience initialisers (for String and ByteBuffer) is an issue either. I'd rather reuse existing protocols if possible but don't feel too strongly.

blindspotbounty · 2023-07-26T14:07:08Z

Sources/SwiftKafka/RDKafka/RDKafkaClient.swift

-                    keyBytes?.count ?? 0,
-                    UnsafeMutableRawPointer(bitPattern: newMessageID)
-                )
+        let responseCode = try message.value.withUnsafeRawBufferPointer { valueBuffer in


Can we create the following method in message, please:

func withUnsafeKeyValueRawBufferPointer(_ closure: (Buffer?, Buffer)) { let responseCode = try value.withUnsafeRawBufferPointer { valueBuffer in if let key { key.withUnsafeRawBufferPointer { keyBuffer in closure(keyBuffer, valueBuffer) } } else { closure(nil, valueBuffer) } } }

Then we could avoid call key.withUnsafeRawBufferPointer for every topic. Especially, it might make sense for custom types.

Ou, sorry, I see that it is only for one topic.

Motivation: We want to allow our users to feed arbitrary data as `key` and `value` of a `KafkaProducerMessage`. Modifications: * add new `public protocol KafkaBuffer` which asks implementing types to provide safe access to an `UnsafeMutableRawBufferPointer` containing the data * implement `KafkaBuffer` in `String` * implement `KafkaBuffer` in `ByteBuffer` * update tests

Modifications: * remove duplicate docc comments for `KafkaBuffer` adopters * spell out `K` and `V` as `Key` and `Value` * `String+KafkaBuffer`: only convert `String` to `ByteBuffer` on slow path (no contiguous bytes availble) * make `Never` conform to `KafkaBuffer` * `KafkaProducerMessage`: make `key` optional

Modifications: * change `KafkaBuffer` `protocol` into `KafkaContiguousBytes` type (similar to `SwiftProtobuf/SwiftProtobufContiguousBytes`) * make `Array where Array.Element == UInt8` conform to `KafkaContiguousBytes`

FranzBusch · 2023-07-28T08:19:40Z

Sources/SwiftKafka/Data/String+KafkaContiguousBytes.swift

+            return try ByteBuffer(string: self).withUnsafeReadableBytes { unsafeRawBufferPointer in
+                try body(unsafeRawBufferPointer)
+            }


NIT: We can just call our other conformance here

Suggested change

return try ByteBuffer(string: self).withUnsafeReadableBytes { unsafeRawBufferPointer in

try body(unsafeRawBufferPointer)

}

return try ByteBuffer(string: self).withUnsafeBytes(body)

FranzBusch · 2023-07-28T08:21:33Z

Sources/SwiftKafka/KafkaProducerMessage.swift


-    /// Create a new `KafkaProducerMessage` with a `String` key and value
+extension KafkaProducerMessage where Key == Never {
+    /// Create a new `KafkaProducerMessage` with a `ByteBuffer` key and value


This looks outdated

Modifications: * fix outdated `KafkaProducerMessage` documentation * simplify `String+KafkaContiguousBytes`

felixschlegel commented Jul 21, 2023

View reviewed changes

felixschlegel force-pushed the fs-kafka-data-type branch from cd1a40e to 5082120 Compare July 21, 2023 16:42

FranzBusch reviewed Jul 24, 2023

View reviewed changes

glbrntt reviewed Jul 24, 2023

View reviewed changes

felixschlegel requested review from FranzBusch and glbrntt July 24, 2023 16:11

blindspotbounty reviewed Jul 26, 2023

View reviewed changes

blindspotbounty approved these changes Jul 26, 2023

View reviewed changes

felixschlegel force-pushed the fs-kafka-data-type branch 2 times, most recently from 900e8cc to 0f21615 Compare July 26, 2023 15:26

felixschlegel requested a review from blindspotbounty July 26, 2023 15:27

felixschlegel added 4 commits July 27, 2023 19:14

KafkaContiguousBytes type

a594a7e

Modifications: * change `KafkaBuffer` `protocol` into `KafkaContiguousBytes` type (similar to `SwiftProtobuf/SwiftProtobufContiguousBytes`) * make `Array where Array.Element == UInt8` conform to `KafkaContiguousBytes`

Fix post-rebase issue

09658d9

felixschlegel force-pushed the fs-kafka-data-type branch from 0f21615 to 09658d9 Compare July 27, 2023 18:14

FranzBusch reviewed Jul 28, 2023

View reviewed changes

Changes Franz

b83a49c

Modifications: * fix outdated `KafkaProducerMessage` documentation * simplify `String+KafkaContiguousBytes`

felixschlegel requested a review from FranzBusch July 28, 2023 08:29

FranzBusch approved these changes Jul 28, 2023

View reviewed changes

FranzBusch merged commit a3df8d8 into swift-server:main Jul 28, 2023

FranzBusch deleted the fs-kafka-data-type branch July 28, 2023 08:39

KafkaProducerMessage: generic key & value #95

KafkaProducerMessage: generic key & value #95

Uh oh!

Conversation

felixschlegel commented Jul 21, 2023

Motivation:

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!

blindspotbounty Jul 26, 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!

Uh oh!

`KafkaProducerMessage`: generic key & value #95

`KafkaProducerMessage`: generic key & value #95

blindspotbounty Jul 26, 2023 •

edited

Loading