Repository: camel Updated Branches: refs/heads/master 13b39177c -> aa9314f26
Added deserializers to class loader resolver Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/aa9314f2 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/aa9314f2 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/aa9314f2 Branch: refs/heads/master Commit: aa9314f26533ca96ef0cd743c69fafe73c0dff1b Parents: 13b3917 Author: John Moon <[email protected]> Authored: Tue Jun 28 15:46:03 2016 -0400 Committer: Claus Ibsen <[email protected]> Committed: Wed Jun 29 07:54:32 2016 +0200 ---------------------------------------------------------------------- .../org/apache/camel/component/kafka/KafkaEndpoint.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/aa9314f2/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java ---------------------------------------------------------------------- diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java index f640bdb..74aa883 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java @@ -32,13 +32,16 @@ import org.apache.camel.spi.ClassResolver; import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.util.CastUtils; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + /** * The kafka component allows messages to be sent to (or consumed from) Apache Kafka brokers. */ @@ -129,8 +132,10 @@ public class KafkaEndpoint extends DefaultEndpoint implements MultipleConsumersS try { if (getCamelContext() != null) { ClassResolver resolver = getCamelContext().getClassResolver(); - replaceWithClass(props, "key.serializer", resolver, Serializer.class); - replaceWithClass(props, "value.serializer", resolver, Serializer.class); + replaceWithClass(props, ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, resolver, Serializer.class); + replaceWithClass(props, ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, resolver, Serializer.class); + replaceWithClass(props, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, resolver, Deserializer.class); + replaceWithClass(props, ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, resolver, Deserializer.class); try { //doesn't exist in old version of Kafka client so detect and only call the method if
