4

使用 Apache Camel（版本 2.19.1）的 kafka 组件时出错，我只是想打印主题中的传入消息，我的管道是如此组合：

...
context.addRoutes(new RouteBuilder() {
            public void configure() {
                from("kafka://localhost:9092?topic=test&groupId=testing")
               .to("stream:out");
  context.start();
    }
}

尝试在端点中使用和不使用“//”。

我得到的是：

Exception in thread "main" org.apache.camel.FailedToCreateRouteException: Failed to create route route1: Route(route1)[[From[kafka://localhost:9092?topic=test&groupI... because of Brokers must be configured
at org.apache.camel.impl.RouteService.warmUp(RouteService.java:147)
at org.apache.camel.impl.DefaultCamelContext.doWarmUpRoutes(DefaultCamelContext.java:3762)
at org.apache.camel.impl.DefaultCamelContext.safelyStartRouteServices(DefaultCamelContext.java:3669)
at org.apache.camel.impl.DefaultCamelContext.doStartOrResumeRoutes(DefaultCamelContext.java:3455)
at org.apache.camel.impl.DefaultCamelContext.doStartCamel(DefaultCamelContext.java:3309)
at org.apache.camel.impl.DefaultCamelContext.access$000(DefaultCamelContext.java:202)
at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:3093)
at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:3089)
at org.apache.camel.impl.DefaultCamelContext.doWithDefinedClassLoader(DefaultCamelContext.java:3112)
at org.apache.camel.impl.DefaultCamelContext.doStart(DefaultCamelContext.java:3089)
at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:3026)
at org.apache.camel.MainApp.main(MainApp.java:60)
Caused by: java.lang.IllegalArgumentException: Brokers must be configured
at org.apache.camel.component.kafka.KafkaConsumer.<init>(KafkaConsumer.java:62)
at org.apache.camel.component.kafka.KafkaEndpoint.createConsumer(KafkaEndpoint.java:76)
at org.apache.camel.impl.EventDrivenConsumerRoute.addServices(EventDrivenConsumerRoute.java:69)
at org.apache.camel.impl.DefaultRoute.onStartingServices(DefaultRoute.java:103)
at org.apache.camel.impl.RouteService.doWarmUp(RouteService.java:172)
at org.apache.camel.impl.RouteService.warmUp(RouteService.java:145)
... 12 more

Process finished with exit code 1

我正在尝试解决这个问题，但我真的不明白问题出在哪里，我的 kafka 集群是一个单独的代理，并且一切都已启动并正在运行（动物园管理员和服务器），请寻求帮助

4

2 回答 2

2

添加brokers=localhost:9092到消费者 uri。

于 2017-09-01T14:19:51.477 回答
1

查看此示例，URL 的第一部分是topic，然后作为参数您可以传递brokers. 所以官方文档对我来说似乎有点误导。

from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}"
                    + "&maxPollRecords={{consumer.maxPollRecords}}"
                    + "&consumersCount={{consumer.consumersCount}}"
                    + "&seekTo={{consumer.seekTo}}"
                    + "&groupId={{consumer.group}}")
                    .routeId("FromKafka")
                .log("${body}");

但作为一般建议：Camel 是开源的，因此您可以随时查看 github 上的代码和示例。您还可以找到您在此处发布的堆栈跟踪的那些行，然后追踪丢失的内容。

于 2017-09-01T15:04:21.533 回答