Kafka Message Header API #120
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
felixschlegel
force-pushed
the
fs-kafka-message-header
branch
from
2dbf00b to
bb8244c
Compare
felixschlegel
commented
Aug 22, 2023
Sources/Kafka/KafkaHeader.swift
Outdated
| public var key: String | ||
|
|
||
| /// The value associated with the header. | ||
| public var value: ByteBuffer? // TODO: KafkaContiguousBytes |
There was a problem hiding this comment.
Ideally, we would do the same KafkaContiguousBytes dance here as in KafkaProducerMessage.
However, the problem is that KafkaHeader is used in an array, which means the entire array would have to be generic over one specific KafkaContiguousBytes type which feels wrong too.
There was a problem hiding this comment.
I am fine to leave this with ByteBuffer for now. Restricting the Array to single type isn't great.
felixschlegel
commented
Aug 22, 2023
| self._internal[index].vtype = RD_KAFKA_VTYPE_HEADER | ||
|
|
||
| // Copy name C String (strdup) as we do not own the headers: [KafkaHeader] array | ||
| guard let headerNameCPointer = header.key.withCString(strdup) else { |
There was a problem hiding this comment.
I am not really happy with doing the strdup here. However, we need to somehow need to pass a char * pointer which we get from a header entry in the headers: [KafkaHeader] array (hence accessing the pointer inside of the closure is weird).
I had a couple of ideas to reduce copying here:
- write C glue code that takes the
headers: [KafkaHeader]array and does essentially whatRDKakfaUnsafeProducerMessagedoes, just in a more enclosed way - use copy-on-write / a String pool to avoid unnecessary copies of the String pointer (if a message header name occurs more than once we can avoid the additional copy with our String pool)
felixschlegel
commented
Aug 22, 2023
| let valueCopyBuffer = UnsafeMutableRawPointer.allocate(byteCount: value.readableBytes, alignment: 1) | ||
| value.withUnsafeReadableBytes { valueBuffer in | ||
| if let baseAddress = valueBuffer.baseAddress { | ||
| valueCopyBuffer.copyMemory(from: baseAddress, byteCount: valueBuffer.count) |
There was a problem hiding this comment.
Also want to avoid copying here, see comment about strdup
felixschlegel
force-pushed
the
fs-kafka-message-header
branch
2 times, most recently
from
48515b3 to
2f158e3
Compare
FranzBusch
reviewed
Aug 24, 2023
Comment on lines
277
to
283
| if kafkaHeaders.isEmpty { | ||
| // Base case: we have read all kafkaHeaders and now invoke the accessor closure | ||
| // that can safely access the pointers in cHeaders | ||
| return try body(cHeaders) | ||
| } else { | ||
| guard let kafkaHeader = kafkaHeaders.popLast() else { | ||
| fatalError("kafkaHeaders should not be nil") |
There was a problem hiding this comment.
Can we swap this around so that we only guard. Then we don't need the fatalError at all
felixschlegel
force-pushed
the
fs-kafka-message-header
branch
from
2f158e3 to
b393742
Compare
Motivation: Be able to attach headers to a `KafkaProducerMessage` and read out headers attached to a `KafkaConsumerMessage`. Modifications: * create new type `struct KafkaHeader` representing a key-value pair of `String` key and `ByteBuffer` value * add property `headers: [KafkaHeader]` to `KafkaConsumerMessage` and `KafkaConsumerMessage` * use `rd_kafka_produceva` (varidadic arguments) to produce messages as `rd_kafka_produce` did not support setting message headers * create helper class `RDKafkaUnsafeProducerMessage` that helps configuring the varidadic argument array for `rd_kafka_produceva` * add new test asserting that both producing and consuming messages with message headers works
Modifications: * no copying of `KafkaProducerMessage` headers and values -> build scoped accessor helper that recursively accesses all underlying pointers of the `KafkaProducerMessage`'s `headers: [KafkaHeader]` * only use `rd_kafka_produceva` when `message.headers.isEmpty == false`
felixschlegel
force-pushed
the
fs-kafka-message-header
branch
from
b393742 to
bb403b2
Compare
FranzBusch
approved these changes
Aug 24, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation:
Be able to attach headers to a
KafkaProducerMessageand read outheaders attached to a
KafkaConsumerMessage.Modifications:
struct KafkaHeaderrepresenting a key-value pair ofStringkey andByteBuffervalueheaders: [KafkaHeader]toKafkaConsumerMessageandKafkaConsumerMessagerd_kafka_produceva(varidadic arguments) to produce messages asrd_kafka_producedid not support setting message headersRDKafkaUnsafeProducerMessagethat helpsconfiguring the varidadic argument array for
rd_kafka_producevamessage headers works