我面临着
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.
从用 Scala 编写的命令行运行以下 kafka 消费者 API 时出错。可能是什么问题?
object KafkaAggregateConsumerApp extends App{
try {
val properties: Properties = new Properties()
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "0:9092")
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer")
properties.put("group.id", "console-consumer-myapp")
val consumerApp = new KafkaConsumer[String, Int](properties)
consumerApp.subscribe(Pattern.compile("kafkaaggregationsource1"))
try {
while (true) {
val consumerRecord: ConsumerRecords[String, Int] = consumerApp.poll(Duration.ofMinutes(10))
consumerRecord.forEach((each) => println(each.key() + " " + each.value()))
}
} finally {
consumerApp.close()
}
}
catch{
case e: Exception => e.printStackTrace()
}
}
最佳答案
在 source code我可以看到这个 InvalidGroupIdException
groupId
时抛出是 null
,
private void maybeThrowInvalidGroupIdException() { if (groupId == null) throw new InvalidGroupIdException("To use the group management or offset commit APIs, you must " + "provide a valid " + ConsumerConfig.GROUP_ID_CONFIG + " in the consumer configuration."); }
我建议使用
ConsumerConfig.GROUP_ID_CONFIG
设置组 ID而不是 "group.id"
:properties.put(ConsumerConfig.GROUP_ID_CONFIG, "console-consumer-myapp")
关于scala - InvalidGroupIdException : in Kafka scala consumer program even after setting group. ID,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/56202988/