Apache camel-apache kafka集成

Question

我正在学习如何将kafka与apache camel集成，我遇到了以下错误。我在c：/inbox文件夹中创建了一个文件，并希望使用kafka Consumer使用其中的文本。我使用的是apache Camel3.1.0版本。下面是我的代码

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);
}
}

下面是我得到的错误

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```

Answer 1

错误stacktrace表示您的Kafka使用者URI无效（在stacktrace的底部）。而且确实是。

正确的形式是kafka:[topicname]？[options]（检查Camel-Kafka文档）

因此，当我查看您的URI时，可能应该是

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

您的URI存在以下问题：

• 它包含两次Kafka:[topicname]what is invalid
• Kafka:[topicname]之一是Kafka:[brokers]，删除它
• 分号(；)代替&来分隔选项
• 旧版本骆驼-Kafka的动物园管理员选项，删除它们

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