Acknowledgement.acknowledge()在spring-kafka @KafkaListener中抛出异常

Question

当我将enable.auto.commit设置为false并尝试使用基于注释的spring-kafka @KafkaListener手动提交偏移量时,我得到一个org.springframework.kafka.listener.ListenerExecutionFailedException:无法使用传入消息调用侦听器方法

我有一个非常简单的代码如下:

@KafkaListener(id = "someid", topics = "${demo.topic}", containerFactory = "someContainerFactory")
public void listenFooGroup(String message, Acknowledgement ack) {
    System.out.println("Received Messasge in group 'foo': " + message);

    // TODO: Do something with the message
}

当我从制作人发送消息时,我得到以下异常:

org.springframework.kafka.listener.ListenerExecutionFailedException:无法使用传入消息调用侦听器方法.

端点处理程序详细信息

方法[public void com.****.*****.*******.KafkaMessageListener.listenFooGroup(java.lang.String,org.springframework.kafka.support.Acknowledgment)]

Bean [com.****.*****.*******.KafkaMessageListener@5856dbe4]; 嵌套异常是org.springframework.messaging.converter.MessageConversionException:无法处理消息; 嵌套异常是org.springframework.messaging.converter.MessageConversionException:无法从[java.lang.String]转换为[org.springframework.kafka.support.Acknowledgment]以获取GenericMessage [payload = test,headers = {kafka_offset = 57,kafka_receivedMessageKey = null,kafka_receivedPartitionId = 0,kafka_receivedTopic = demotopic}],failedMessage = GenericMessage [payload = test,headers = {kafka_offset = 57,kafka_receivedMessageKey = null,kafka_receivedPartitionId = 0,kafka_receivedTopic = demotopic}]

请帮忙.TIA.

Answer 1

您必须将容器工厂的containerPropertiesackMode设置为MANUAL或MANUAL_IMMEDIATE获取Acknowledgment对象.

对于其他ack模式,容器负责提交偏移量.

factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE)

或者....ackMode如果使用Spring Boot,则设置属性