InvalidGroupIdException：在 Kafka scala 消费者程序中，即使设置 group.id 后也是如此

Question

我面临着

org.apache.kafka.common.errors.InvalidGroupIdException：要使用组管理或偏移提交 API，您必须在使用者配置中提供有效的 group.id。

从用 scala 编写的命令行运行以下 kafka Consumer 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

在源代码中我可以看到这是在isInvalidGroupIdException时抛出的，groupIdnull

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