Spring-Kafka：如何在SmartLifeCycle的stop()中正确提交Kafka最新偏移量

针对你遇到的问题，直接持有全局Consumer实例并在stop线程调用提交操作会触发线程安全警告——因为Kafka Consumer本身不是线程安全的，只能在其所属的消费线程中操作。以下是正确的实现方式：

1. 配置手动提交模式 #

首先确保@KafkaListener使用手动提交模式，可在注解或容器工厂中指定：

@KafkaListener(
    topics = "your-topic",
    containerFactory = "kafkaListenerContainerFactory",
    ackMode = "MANUAL"
)

对应的容器工厂配置：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    // 启用手动提交模式
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
    return factory;
}

2. 线程安全维护已处理偏移量 #

创建线程安全的存储，记录每个分区已处理完成的最新偏移量（注意：提交的偏移量是下一个要消费的位置，即当前消息offset+1）：

@Component
public class OffsetTracker {
    private final ConcurrentMap<TopicPartition, OffsetAndMetadata> processedOffsets = new ConcurrentHashMap<>();

    public void updateOffset(ConsumerRecord<?, ?> record) {
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        processedOffsets.put(tp, new OffsetAndMetadata(record.offset() + 1));
    }

    public Map<TopicPartition, OffsetAndMetadata> getProcessedOffsets() {
        return new HashMap<>(processedOffsets);
    }
}

在@KafkaListener方法中，消息处理完成后更新偏移量：

@KafkaListener(...)
public void listen(ConsumerRecord<String, String> record) {
    // 业务处理逻辑
    processMessage(record);
    // 更新已处理偏移量
    offsetTracker.updateOffset(record);
}

3. 实现SmartLifecycle安全提交偏移量 #

注入KafkaListenerEndpointRegistry和OffsetTracker，在stop方法中通过容器的doWithConsumer方法提交偏移量——该方法会在Consumer所属的消费线程中执行操作，彻底避免线程安全问题：

@Component
public class KafkaShutdownHandler implements SmartLifecycle {
    private final KafkaListenerEndpointRegistry registry;
    private final OffsetTracker offsetTracker;
    private boolean running = false;

    public KafkaShutdownHandler(KafkaListenerEndpointRegistry registry, OffsetTracker offsetTracker) {
        this.registry = registry;
        this.offsetTracker = offsetTracker;
    }

    @Override
    public void start() {
        running = true;
    }

    @Override
    public void stop() {
        // 1. 暂停所有监听器容器，阻止新消息消费
        registry.getListenerContainers().forEach(MessageListenerContainer::pause);

        // 2. 遍历每个容器，在消费线程中提交对应分区的偏移量
        registry.getListenerContainers().forEach(container -> {
            container.doWithConsumer(consumer -> {
                Map<TopicPartition, OffsetAndMetadata> allOffsets = offsetTracker.getProcessedOffsets();
                // 只提交当前容器负责的分区偏移量
                Map<TopicPartition, OffsetAndMetadata> containerOffsets = allOffsets.entrySet().stream()
                        .filter(entry -> container.getAssignedPartitions().contains(entry.getKey()))
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
                if (!containerOffsets.isEmpty()) {
                    consumer.commitSync(containerOffsets);
                }
            });
        });

        // 3. 停止所有容器
        registry.stop();
        running = false;
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}

关键说明 #

• 避免跨线程操作Consumer：doWithConsumer方法将提交逻辑委托给Consumer的原生消费线程执行，完全符合Kafka Consumer的线程安全要求，不会触发警告。
• 先暂停再提交：暂停容器可确保提交过程中没有新消息被消费，避免偏移量提交后仍有未处理消息残留。
• 分区匹配提交：仅提交当前容器负责的分区偏移量，避免跨容器的偏移量提交错误。

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