Spring Boot BatchAcknowledgeingMessageListener在逗号上拆分消息

Question

我有一个Spring Boot应用程序，其中一个Kafka侦听器实现了BatchAcknowledgeingMessageListener

生成记录的代码如下所示：

this.kafkaTemplate.send("myTopic", "12345", "{\"OrderID\": \"12345\"}, \"OrderDate\": \"2021-06-01T12:13:16Z\"");

Kafka的配置看起来是这样的（这仍处于使用Testcontainers的集成测试阶段，因此制作人正在制作与消费者正在听的主题相同的内容）：

spring:
  kafka:
    listener:
      ack-mode: manual-immediate
      concurrency: 1
    consumer:
      bootstrap-servers: localhost:9093
      enable-auto-commit: false
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 10
      topic: myTopic
    producer:
      bootstrap-servers: localhost:9093
      client-id: my-client
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      topic: myTopic

最后，消费者逻辑：

@KafkaListener(topics = "${spring.kafka.consumer.topic}")
public void consumeMessages(final List<ConsumerRecord<String, String>> records, final Acknowledgment ack) throws IOException {
  // This line fails with ClassCastException: "Can't cast String to ConsumerRecord"
  // for (final ConsumerRecord<String, String> record : records) {

  for (final Object record : records) {
    log.debug("Record: {}", record);
  }
  ...
}

此示例的调试输出为：

[LOG HEADER]: Record: {"OrderID": "12345"
[LOG HEADER]: Record: "OrderDate": "2021-06-01T12:13:16Z"}

正如你所看到的，这条消息正在用逗号分割，我收到的一条消息中包含多条消息。这显然是失败的，但我不明白为什么我不只是得到一条消费记录

Answer 1

原来问题很简单。我错过了spring.kafka.listener.type配置参数表。在这里找到了答案：

Answer 2

您缺少侦听器类型配置，因此默认转换服务会看到您想要一个列表，并用逗号分割字符串。

spring:
  kafka:
    listener:
      ack-mode: manual-immediate
      concurrency: 1
      type: batch
    consumer:
...

添加类型：批处理告诉框架您需要完整的记录批处理。