带Manual_Inmediate ack的Spring靴Kafka

Question

  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

@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();
    }
}