Apache camel-apache kafka integration

Question

I am learning how to integrate kafka with apache camel and i encountered the following error. Any help will be appreciated.I have a file created inside C:/inbox folder and want to consume the text in it using kafka consumer.I am using version 3.1.0 of apache camel.Below is my code

package com.javainuse;

import org.apache.camel.builder.RouteBuilder;

public class SimpleRouteBuilder extends RouteBuilder {

@Override
public void configure() throws Exception {

    String topicName = "test123";
    String kafkaServer = "kafka:localhost:9092";
    String zooKeeperHost = "zookeeperHost=localhost&zookeeperPort=2181";
    String serializerClass = "serializerClass=kafka.serializer.StringEncoder";
    String toKafka = "kafka:localhost:9092;kafka:test123?brokers=localhost:9092;zookeeperHost=localhost;zookeeperPort=2181;groupId=group1";

    from("file:C:/inbox?noop=true").to(toKafka);
}
}

And below is the error I am getting

SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
org.apache.camel.FailedToStartRouteException: Failed to start route route1 because of Route(route1)[From[file:C:/inbox?noop=true] -> [To[kafka:loc...
    at org.apache.camel.impl.engine.BaseRouteService.warmUp(BaseRouteService.java:133)
    at org.apache.camel.impl.engine.AbstractCamelContext.doWarmUpRoutes(AbstractCamelContext.java:3246)
    at org.apache.camel.impl.engine.AbstractCamelContext.safelyStartRouteServices(AbstractCamelContext.java:3139)
    at org.apache.camel.impl.engine.AbstractCamelContext.doStartOrResumeRoutes(AbstractCamelContext.java:2925)
    at org.apache.camel.impl.engine.AbstractCamelContext.doStartCamel(AbstractCamelContext.java:2725)
    at org.apache.camel.impl.engine.AbstractCamelContext.lambda$doStart$2(AbstractCamelContext.java:2527)
    at org.apache.camel.impl.engine.AbstractCamelContext.doWithDefinedClassLoader(AbstractCamelContext.java:2544)
    at org.apache.camel.impl.engine.AbstractCamelContext.doStart(AbstractCamelContext.java:2525)
    at org.apache.camel.support.service.ServiceSupport.start(ServiceSupport.java:121)
    at org.apache.camel.impl.engine.AbstractCamelContext.start(AbstractCamelContext.java:2421)
    at com.javainuse.MainApp.main(MainApp.java:12)
Caused by: org.apache.camel.RuntimeCamelException: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
    at org.apache.camel.RuntimeCamelException.wrapRuntimeCamelException(RuntimeCamelException.java:52)
    at org.apache.camel.support.ChildServiceSupport.start(ChildServiceSupport.java:67)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:70)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:87)
    at org.apache.camel.processor.channel.DefaultChannel.doStart(DefaultChannel.java:144)
    at org.apache.camel.support.service.ServiceSupport.start(ServiceSupport.java:121)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:70)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:73)
    at org.apache.camel.processor.Pipeline.doStart(Pipeline.java:154)
    at org.apache.camel.support.service.ServiceSupport.start(ServiceSupport.java:121)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:70)
    at org.apache.camel.support.processor.DelegateAsyncProcessor.doStart(DelegateAsyncProcessor.java:78)
    at org.apache.camel.support.service.ServiceSupport.start(ServiceSupport.java:121)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:70)
    at org.apache.camel.impl.engine.BaseRouteService.startChildService(BaseRouteService.java:339)
    at org.apache.camel.impl.engine.BaseRouteService.doWarmUp(BaseRouteService.java:189)
    at org.apache.camel.impl.engine.BaseRouteService.warmUp(BaseRouteService.java:131)
    ... 10 more
Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:432)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
    at org.apache.camel.component.kafka.KafkaProducer.doStart(KafkaProducer.java:119)
    at org.apache.camel.support.service.ServiceSupport.start(ServiceSupport.java:121)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:70)
    at org.apache.camel.impl.engine.AbstractCamelContext.internalAddService(AbstractCamelContext.java:1455)
    at org.apache.camel.impl.engine.AbstractCamelContext.addService(AbstractCamelContext.java:1391)
    at org.apache.camel.processor.SendProcessor.doStart(SendProcessor.java:240)
    at org.apache.camel.support.service.ServiceSupport.start(ServiceSupport.java:121)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:70)
    at org.apache.camel.support.service.ServiceHelper.startService(ServiceHelper.java:87)
    at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler.doStart(RedeliveryErrorHandler.java:1454)
    at org.apache.camel.support.ChildServiceSupport.start(ChildServiceSupport.java:60)
    ... 25 more
Caused by: org.apache.kafka.common.config.ConfigException: Invalid url in bootstrap.servers: localhost:9092;zookeeperHost=localhost;zookeeperPort=2181;groupId=group1
    at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:58)
    at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:47)
    at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:407)
    ... 37 more

Process finished with exit code 0```

burki burki · Accepted Answer · 2020-07-01T05:27:29

The error stacktrace says that your Kafka consumer URI is invalid (at the bottom of the stacktrace). And it is indeed.

The correct form is kafka:[topicname]?[options] (check Camel-Kafka docs)

So when I look at your URI it should probably be

kafka:test123?brokers=localhost:9092&groupId=group1

Your URI has the following problems:

It contains 2 times kafka:[topicname] what is invalid
One of the kafka:[topicname] is kafka:[brokers], remove it
Semicolons (;) instead of & to delimit options
Zookeeper options for old versions of camel-kafka, remove them

By the way: The line SLF4J: Defaulting to no-operation (NOP) logger implementation on top of your stacktrace says that you use SLF4J logging interface, but you have no implementation added to your project.

If you use Maven, you can add the following dependency to add the SLF4J API as well as Logback as implementation to your project.

<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
</dependency>

Apache camel-apache kafka integration

1 Answers