Kafka transport does not respect the partitioning extension

[bsideup]

https://github.com/cloudevents/spec/blob/master/kafka-protocol-binding.md#31-key-attribute

https://github.com/cloudevents/spec/blob/v1.0/extensions/partitioning.md

[fabiojose]

We don't have implemented this extension yet. PRs are welcome!

[slinkydeveloper]

Since now the sdk provides only a value *Serializer and not a producer anymore, I'm not sure where this functionality should live... Maybe in a key serializer that gets an event and extracts the key?

BTW for reference this is how I implemented it in sdk-go: https://github.com/cloudevents/sdk-go/pull/437/files#diff-fc52ba8f8c260341995925b8297d6144R24 . Basically the message writer (in this sdk is mapped to the concept of visitors, so KafkaProducerMessageVisitorImpl and KafkaSerializerMessageVisitorImpl) tries to see if there is partitionKey in the message. If there is, it uses it as key of the record. I think this is feasible and trivial to implement for KafkaProducerMessageVisitorImpl (aka in case the user invokes manually the unmarshalling process), but for the CloudEventSerializer I'm not sure it makes sense, since that serializer works only with headers and body of the message.

[bluemonk3y]

An interceptor is required. The serializer won't have access to headers or the key and value - whereas the interceptors do. I raised a ticket for this previously.

https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/producer/ProducerRecord.html

Regards,
Neil.

[bluemonk3y]

@slinkydeveloper - implementing something that does not support general producer and consumer use means the support for CloudEvents becomes limited. For example - Connectors, Stream Processors cannot interact natively with the Record. The correct approach is to use an interceptor and remove any code the tries to replace or reimplement Producers and Consumers. At that point an Interceptor is required in order to 1) determine the partition key 2) to write to the KafkaHeader.

[slinkydeveloper]

@bluemonk3y that sounds interesting, are you willing to PR a poc (even not working) for that based on the actual code? I don't have enough expertise on Interceptors to understand if we really need them or not. I'll give a look next week to Interceptors...
I would love to keep the Serializer/Deserializer if we can because they're damn simple to use (and afaik they work with streams and connectors https://docs.confluent.io/current/streams/developer-guide/datatypes.html)

[bluemonk3y]

Hi Francesco, I would love help however I don't have any spare cycles working - Im working on another OpenSource project. I can assist in PR's etc but otherwise, I'm really short on time. Regards, Neil

…

On Fri, 29 May 2020 at 14:40, Francesco Guardiani ***@***.***> wrote: @bluemonk3y <https://github.com/bluemonk3y> that sounds interesting, are you willing to PR a poc for that based on the actual code? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#94 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ACQLJJS5Y4Q4SUNSZM5RQQLRT63MXANCNFSM4KQ4IVOA> .

[slinkydeveloper]

I think we can provide an interceptor that assigns the partitionKey and then serialization is done as usual by CloudEventSerializer. I'm working on it