scala - InvalidGroupIdException : in Kafka scala consumer program even after setting group. ID

Question

我面临着

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

Answer 1

在 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")