Spring-Kafka中@KafkaHandler方法如何实现手动提交（Manual Commit）？

当然可以实现手动提交！你遇到的Acknowledgement对象无效的问题，大概率是因为容器的提交模式没有配置为手动，或者相关配置缺失导致的。下面我会一步步告诉你怎么解决这个问题：

1. 核心前提：配置容器工厂为手动提交模式 #

Spring Kafka默认是自动提交偏移量的，所以要让Acknowledgement生效，必须先把容器的AckMode设置为MANUAL或者MANUAL_IMMEDIATE。

你需要创建一个ConcurrentKafkaListenerContainerFactory的Bean，指定对应的提交模式：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.ContainerProperties;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 设置手动提交模式：MANUAL需调用acknowledge()才提交；MANUAL_IMMEDIATE会立即提交偏移量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka集群地址");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "mygroup");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 配置JSON反序列化，适配你的MyEvent类型
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.your.package"); // 替换为MyEvent所在的包路径
        // 明确关闭自动提交（Spring会在设置MANUAL模式时自动关闭，但显式设置更清晰）
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new DefaultKafkaConsumerFactory<>(props);
    }
}

2. 调整@KafkaHandler方法的参数 #

你的原有代码结构是对的，还可以选择更灵活的ConsumerAwareAcknowledgement（它继承了Acknowledgement，还能获取Kafka Consumer实例，方便做更多底层操作）：

import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.ConsumerAwareAcknowledgment;
import org.springframework.stereotype.Service;

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

    @KafkaHandler
    public void consumeEvent(MyEvent event, ConsumerAwareAcknowledgment ack) throws Exception {
        try {
            // 执行你的业务处理逻辑
            System.out.println("处理事件：" + event.toString());
            // 手动提交偏移量
            ack.acknowledge();
        } catch (Exception e) {
            // 异常时可根据业务决定是否重试、跳过或放弃提交
            e.printStackTrace();
        }
    }
}

关键注意点 #

• 如果你的@KafkaListener指定了自定义容器工厂，一定要确保该工厂的AckMode是手动模式（比如上面代码中通过containerFactory参数绑定配置好的工厂）
• 使用MANUAL模式时，只有调用acknowledge()方法偏移量才会被提交；如果方法抛出异常，Spring不会自动提交偏移量，你可以在catch块里做针对性处理
• 不要同时开启自动提交（ENABLE_AUTO_COMMIT_CONFIG=true）和手动提交模式，这会导致偏移量提交逻辑混乱

按以上配置调整后，你就能在@KafkaHandler方法中拿到有效的Acknowledgement对象，顺利实现手动提交偏移量了。

内容的提问来源于stack exchange，提问作者Jungan