Spring Boot Kafka with MANUAL_IMMEDIATE ack

Question

我有一个Springboot应用程序。使用此配置：

kafka:
consumer:
listener:
ack-mode: MANUAL_IMMEDIATE

广告这个消费者：

@KafkaListener(topics = "test", groupId = "group_id")
public void consume(String message, Acknowledgment ack) throws IOException {
ack.acknowledge();
}

但我在收到一条消息时出现了这个错误：

org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [org.springframework.kafka.support.Acknowledgment] for GenericMessage 

Answer 1

您的YAML似乎没有被读取；我刚刚测试了

更正

您有无效的YAML-删除consumer元素。

spring:
kafka:
consumer:
auto-offset-reset: earliest
listener:
ack-mode: MANUAL_IMMEDIATE

而且效果很好；删除ack-mode会出现与您看到的错误相同的错误。

@SpringBootApplication
public class So60929385Application {
public static void main(String[] args) {
SpringApplication.run(So60929385Application.class, args);
}
@KafkaListener(id = "so60929385", topics = "so60929385")
public void listen(String in, Acknowledgment ack) {
System.out.println(in);
ack.acknowledge();
}
@Bean
public ApplicationRunner runner(KafkaTemplate<String, String> template) {
return args -> template.send("so60929385", "foo");
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("so60929385").partitions(1).replicas(1).build();
}
}