Missing message key on outbound using Camel Kafka KafkaConnector

[swierszczu4j]

As title says - I have a problem in which I would like to rewrite kafka record (key + payload) from one topic to another, but no matter if I'm using Sink or Source option of this connector, the outbound topic receives message without a key (key ends up being null), which also means partition ordering guarantee is lost.

My setup:
curl --location --request PUT 'http://localhost:8083/connectors/camel-sink-connector/config' \ --header 'Content-Type: application/json' \ --data-raw '{ "connector.class": "org.apache.camel.kafkaconnector.kafka.CamelKafkaSinkConnector", "tasks.max": "1", "camel.sink.contentLogLevel": "DEBUG", "camel.sink.path.topic": "example.topic.2", "camel.sink.endpoint.brokers": "kafka-1:9092,kafka-2:9092,kafka-3:9092", "topics": "example.topic", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter": "org.apache.kafka.connect.storage.StringConverter", "consumer.override.isolation.level": "read_committed" }'

Using: https://camel.apache.org/camel-kafka-connector/latest/connectors/camel-kafka-kafka-sink-connector.html in version 0.7.1

And while sending message (key, payload) onto inbound topic - example.topic - on outbound - example.topic.2 - I receive (null, payload). Is there a way to adjust this configuration to have desire behavior - key is also being past to an endpoint?

[oscerd]

For the source side you have to specify this configuration option
https://github.com/apache/camel-kafka-connector/blob/master/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java#L70

which is null by default. On the sink side, the key will be in this header https://github.com/apache/camel-kafka-connector/blob/master/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java#L45

[swierszczu4j]

Hey - thanks for the reply! Much appreciated! As I'm new to the Camel itself I must admit that I do not know how to use properties mentioned by U or what to place as a value for this keys in my connector config so to have rewrite of the kafka message key from one topic to another. All I was able to achieve is to have this key as camel specific header, but message key still is null. Do U think preparing a simple repo usecasing what I want to achieve could be of a help to U in sorting out my issue?

[oscerd]

Hey @swierszczu4j absolutely. It would be a good starting point. Thanks for your time.

[swierszczu4j]

Hello - sorry for a delay. Here's a repo with showcase:
https://github.com/swierszczu4j/camel-kafka-kafka-connector-problem

[swierszczu4j]

Hey @oscerd - have U had time to look into the showcase repo? Do U need additional info?

[oscerd]

Hello @swierszczu4j still didn't find the time.

[oscerd]

I'll try to have a look this week.

[oscerd]

@swierszczu4j I opened this PR on your reproducer
https://github.com/swierszczu4j/camel-kafka-kafka-connector-problem/pull/1/files

Basically you need to add a "CamelHeader.kafka.KEY" header with the key value while sending message to the inbound topic.

I know it wasn't clear completely, but it's not easy to generalize the Camel Headers usage in the ckc context.

Please let me know if it works for you and sorry for the long delay.

[swierszczu4j]

@oscerd Thank U very much - indeed it works now! Just out of curiosity, for Source Connector solution would be similar, so to add this config options:
https://github.com/apache/camel-kafka-connector/blob/master/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java#L74
With value "CamelHeader.kafka.KEY", correct?

[oscerd]

Yes, it should work in that way.

[swierszczu4j]

Hey @oscerd - last thing, as I have tried to make it work with Source connector, yet could not have key on the outbound. The test to reproduce is in the repo: https://github.com/swierszczu4j/camel-kafka-kafka-connector-problem/blob/main/src/test/kotlin/com/example/demo/ShowcaseSourceConnectorTest.kt
I followed the last comment, but it's not working. Can U help also with this one, please?