[
https://issues.apache.org/jira/browse/CAMEL-12503?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16470913#comment-16470913
]
ASF GitHub Bot commented on CAMEL-12503:
----------------------------------------
tdanylchuk commented on a change in pull request #2333: CAMEL-12503: added
support for propagating camel headers to kafka and vice versa
URL: https://github.com/apache/camel/pull/2333#discussion_r187415859
##########
File path:
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
##########
@@ -214,15 +222,39 @@ public void remove() {
ProducerRecord record;
if (hasPartitionKey && hasMessageKey) {
- record = new ProducerRecord(topic, partitionKey, key, value);
+ record = new ProducerRecord(topic, partitionKey, null, key, value,
propagatedHeaders);
} else if (hasMessageKey) {
- record = new ProducerRecord(topic, key, value);
+ record = new ProducerRecord(topic, null, null, key, value,
propagatedHeaders);
} else {
- record = new ProducerRecord(topic, value);
+ record = new ProducerRecord(topic, null, null, null, value,
propagatedHeaders);
}
return Collections.singletonList(record).iterator();
}
+ private List<Header> getPropagatedHeaders(Exchange exchange,
HeaderFilterStrategy headerFilterStrategy) {
+ return exchange.getIn().getHeaders().entrySet().stream()
+ .filter(entry ->
!headerFilterStrategy.applyFilterToExternalHeaders(entry.getKey(),
entry.getValue(), exchange))
+ .map(entry -> new RecordHeader(entry.getKey(),
getHeaderValue(entry.getValue())))
+ .collect(Collectors.toList());
+ }
+
+ private byte[] getHeaderValue(Object value) {
Review comment:
I've put most common types, so if you put unsupported one - value will be
null.
The docs of kafka says that we need to bypass `byte[]` , but in camel
headers we have `Object`, so I need somehow to know how to convert provided
object to bytes.
Although I agree, that `Double` also good to have supported.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Kafka component should be able to propagate camel headers to kafka
> ------------------------------------------------------------------
>
> Key: CAMEL-12503
> URL: https://issues.apache.org/jira/browse/CAMEL-12503
> Project: Camel
> Issue Type: New Feature
> Reporter: Taras Danylchuk
> Priority: Major
>
> Since 0.11.0 Kafka support headers, and it would be awesome to have such
> feature available also in camel component.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
