Composant Camel-Kafka ne fonctionne pas pour l'erreur: "en raison de courtiers doivent être configurés"

Question

Vous avez une erreur en utilisant le composant kafka pour Apache Camel (version 2.19.1), j'essaie juste d'imprimer les messages entrants dans le sujet, mon pipeline est ainsi composé:

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

essayé avec et sans "//" dans l'extrémité.

Ce que je suis arrivé est:

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 

Je suis en train de le comprendre, mais je ne comprends vraiment pas ce qui est le problème, mon groupe de kafka est un seul courtier et tout est en cours d'exécution (zookeeper et serveur), ty pour l'aide

Answer 1

Ajouter brokers=localhost:9092 au consommateur uri.

Answer 2

En regardant this example la première partie de l'URL est le topic, puis en tant que paramètre, vous pouvez passer le brokers. Donc, la documentation officielle semble être un peu trompeuse pour moi.

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

Mais comme des conseils généraux: Camel est open source, donc vous pouvez toujours jeter un oeil sur le code et les exemples là GitHub. Vous pouvez également trouver les lignes de la trace de la pile que vous avez posté là, puis traquer ce qui manque.