InvalidGroupIdException для носика Кафки в шторме - PullRequest
Новая
       22

InvalidGroupIdException для носика Кафки в шторме

0 голосов
/ 15 января 2020

Я определил основы c Топология шторма с потребителем носика от Kafka (производитель создан в отдельном модуле Kafka). Однако, когда я запускаю приложение, я получаю эту ошибку: 

java.lang.RuntimeException: org.apache.kafka.common.errors.InvalidGroupIdException: To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.
    at org.apache.storm.utils.Utils$1.run(Utils.java:407) ~[storm-client-2.1.0.jar:2.1.0]
    at java.lang.Thread.run(Thread.java:748) [?:1.8.0_221]
Caused by: org.apache.kafka.common.errors.InvalidGroupIdException: To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.

Как я могу настроить идентификатор группы? Я запускаю Storm локально с версией 2.1.0.

Вот код для топологии: 

val cluster = new LocalCluster()

val bootstrapServers = "localhost:9092"
val brokerHosts = new ZkHosts(bootstrapServers)
val topologyBuilder = new TopologyBuilder()

val spoutConfig = KafkaSpoutConfig.builder(bootstrapServers, "tweets").build()
topologyBuilder.setSpout("kafka_spout", new KafkaSpout(spoutConfig), 1)

val config = new Config()
cluster.submitTopology("kafkaTest", config, topologyBuilder.createTopology())

1 Ответ

2 голосов
/ 15 января 2020

Вы должны использовать setProp(java.lang.String, java.lang.Object) с ConsumerConfig.GROUP_ID_CONFIG, чтобы добавить идентификатор группы потребителей на KafkaSpoutConfig

...