@KafkaHandler中的手动提交（spring kafka）

Question

我正在使用方法级别的@KafkaHandler（类级别的@KafkaListener）消费Kafka事件。

我见过很多例子，其中有一个“acknowledge”参数可用，在这个参数上可以调用“acknowledge（）”方法来提交事件的消耗，但是，当我将它作为参数包含到我的方法中时，我无法填充acknowledge对象。使用KafkaHandler时如何手动提交？有可能吗？

代码示例：

@Service
@KafkaListener(topics = "mytopic", groupId = "mygroup")
public class TestListener {

@KafkaHandler
public void consumeEvent(MyEvent event, Acknowledgement ack) throws Exception {
    //... processing
    ack.acknowledge(); // ack is not available
}

使用SpringBoot和Spring kafka。

Answer 1

必须使用AckMode配置侦听器容器。手动或确认模式。手动_IMMEDIATE获取此功能。

但是，通常最好让容器负责使用AckMode提交偏移量。RECORD或AckMode。BATCH（默认）。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-抵消

编辑

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=MANUAL

@SpringBootApplication
public class So68844554Application {

    public static void main(String[] args) {
        SpringApplication.run(So68844554Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so68844554").partitions(1).replicas(1).build();
    }

}

@Component
@KafkaListener(id =  "so68844554", topics = "so68844554")
class Foo {

    @KafkaHandler
    void listen(String in, Acknowledgment ack) {
        System.out.println(in);
        ack.acknowledge();
    }

}

% kafka-consumer-groups --bootstrap-server localhost:9092 --describe -group so68844554

Consumer group 'so68844554' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
so68844554      so68844554      0          2               2               0               -               -               -