accountledgement.accountled（）在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
}

并且当我从生产者发送消息时,我得到以下异常：

Listener.ExecutionFailedException：无法使用传入消息调用侦听器方法。

终结点处理程序详细信息：

方法[public void com.****.****.******.KafkaMessageListener.ListenFoogroup（java.lang.String,org.SpringFramework.Kafka.Support.Acknowdgment)]

Bean[com.****.*****.******.kafkamessageListener@5856DBE4]；嵌套异常为org.springframework.messaging.converter.messageConversionexception：无法处理消息；嵌套异常为org.springframework.messaging.converter.messageConversionexception：无法将GenericMessage[payload=test,headers={kafka_offset=57,kafka_receivedmessagekey=null,kafka_receivedpartitionid=0,kafka_receivedtopic=demotopic}]、failedmessage=GenericMessage[payload=test,headers={kafka_offset=57,

请帮帮忙。蒂娅。

Answer 1

您必须将容器工厂的containerpropertiesackMode设置为manual或manual_immedial以获取确认对象。

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

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

如果使用Spring Boot，则设置....ackmode属性