Storm 中 Kafka spout 的 InvalidGroupIdException

Question

我已经用来自 Kafka 的 spout 消费者定义了一个基本的 Storm 拓扑（生产者是在 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.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.

如何设置群号？我在本地使用 2.1.0 版本运行ning Storm。

拓扑代码如下：

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())

Answer 1

您应该使用 setProp(java.lang.String, java.lang.Object) 和 ConsumerConfig.GROUP_ID_CONFIG 在 KafkaSpoutConfig

上添加消费者组 ID

Storm 中 Kafka spout 的 InvalidGroupIdException

InvalidGroupIdException for Kafka spout in Storm

scala

apache-kafka

apache-storm