Spring-Kafka确认机制报错：No Acknowledgment available as an argument

问题出现

在spring boot集成kafka时报错，报错信息：

No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.

翻译： Acknowledgment 参数不可用，监听容器需要设置一个手动的 AckMode 才能填充Acknowledgment。

问题代码：

@KafkaListener(topics = PENDING_TOPIC, groupId = KafkaProducer.TOPIC_GROUP1)public void topic_test(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {Optional message = Optional.ofNullable(record.value());if (message.isPresent()) {Object msg = message.get();logger.info("topic.group1 消费了： Topic:" + topic + ",Message:" + msg);ack.acknowledge();}}

这是参考其他人的消费者端的代码，方法里有Acknowledgment 类型参数，说明该参数没有被识别。首先查一下这个参数的含义。

问题分析

Acknowledgment 类型参数的含义：
控制消息的确认机制。在kafka中，消费者提交偏移量时需要确认，执行ack.acknowledge();后代表告诉 Kafka 消费者组，当前的消息已经被成功处理并且可以提交偏移量。这意味着消费者不会再次处理这条消息。

Spring-kafka 提供了多种确认机制，也就是 AckMode。官方文档显示AckMode 有以下几种:
在这里插入图片描述

RECORD: 当监听器处理完记录返回时，提交偏移量.
BATCH: 处理完 poll() 返回的所有记录后，提交偏移量。
TIME: 当poll()返回的所有记录都已处理完毕后，只要超过上次提交后的 ackTime，就提交偏移量。
COUNT: 当poll()返回的所有记录都已处理完毕后，只要上次提交后已收到 ackCount 记录，就提交偏移量。
COUNT_TIME: 与 TIME 和 COUNT 类似，但如果任一条件为真，都会执行提交。
MANUAL: 处理完一批数据后，手动调用 Acknowledgment.acknowledge()方法将offest提交至缓存，之后在下一个poll() 之前用BATCH 方式提交。
MANUAL_IMMEDIATE: 当监听器调用 Acknowledgment.acknowledge() 方法时，立即提交偏移量。

报错原因：

从上面的集中确认机制中可以看到，只有MANUAL 和MANUAL_IMMEDIATE 用到了Acknowledgment 其他都没有。所以需要把AckMode 配置成这两个其中一个才行，而查看配置文件发现我确实没有配。

解决办法：

在application.yaml 中配置ack-mode: manual_immediate 。
在自定义配置AckMode 的时候，首先需要将enable-auto-commit 设置成false 才行，2.3版本之后默认是false，不过使用的默认机制是BATCH，所以还需要将ack-mode设成manual_immediate 来覆盖默认配置。
在这里插入图片描述

spring:kafka:bootstrap-servers: localhost:9092consumer:auto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializer# 值的反序列化方式value-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: geo-friendenable-auto-commit: false # 设置成手动提交auto-commit-interval: 1Sproducer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerlistener:	# 配置监听容器的ackmodeconcurrency: 5ack-mode: manual_immediatemissing-topics-fatal: false