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

【问题标题】：Manual commit in @KafkaHandler (spring-kafka)@KafkaHandler 中的手动提交（spring-kafka）
【发布时间】：2021-08-19 08:38:36
【问题描述】：

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

我已经看到了很多“确认”参数可用的示例，可以调用“确认（）”方法来提交事件的消费，但是，我无法填充确认对象当将其作为我的方法的参数时。使用 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。

【问题讨论】：

标签： apache-kafka spring-kafka

【解决方案1】：

您必须使用AckMode.MANUAL 或AckMode.MANUAL_IMMEDIATE 配置侦听器容器才能获得此功能。

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

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

编辑

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               -               -               -

【讨论】：

谢谢。我已经手动配置了 ackmode 并验证了事件不是 ACK:ed。尽管如此，Acknowledgement 参数并未填充
我刚刚测试没有任何问题；查看编辑；如果您仍然无法弄清楚，请发布 MCRE 来展示您所看到的行为。