Spring Kafka如何在提交偏移量后触发操作且少改现有监听器？

方案1：利用RecordInterceptor的afterRecord方法（消息处理后触发） #

Spring Kafka 2.2及以上版本的RecordInterceptor提供了afterRecord方法，可在消息处理完成（无论成功或失败）后执行自定义逻辑，无需修改原有监听器代码。

• 实现RecordInterceptor接口，重写afterRecord方法：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.listener.RecordInterceptor;

public class PostProcessingRecordInterceptor<K, V> implements RecordInterceptor<K, V> {

    @Override
    public ConsumerRecord<K, V> intercept(ConsumerRecord<K, V> record, Consumer<?, ?> consumer) {
        // 前置操作可选，不需要可直接返回record
        return record;
    }

    @Override
    public void afterRecord(ConsumerRecord<K, V> record, Consumer<?, ?> consumer, Exception exception) {
        // 消息处理后的自定义操作
        if (exception == null) {
            System.out.println("消息处理成功，偏移量：" + record.offset());
            // 此处执行偏移量跟踪或后续业务操作
        } else {
            System.out.println("消息处理失败，偏移量：" + record.offset() + "，异常信息：" + exception.getMessage());
        }
    }
}

• 将拦截器配置到监听容器：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 添加自定义拦截器
        factory.setRecordInterceptor(new PostProcessingRecordInterceptor<>());
        return factory;
    }
}

该方案无需改动原有@KafkaListener实现，仅通过配置即可实现消息处理后的逻辑触发，同时保留自动提交偏移量的机制。

方案2：使用CommitCallback跟踪偏移量提交结果（偏移量提交后触发） #

如果需要严格在偏移量提交成功/失败后执行操作，可以利用Spring Kafka的CommitCallback，它会在自动提交偏移量完成后回调通知。

• 配置容器的提交回调：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.CommitCallback;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.TopicPartitionOffset;

import java.util.Collection;

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        // 获取容器配置，设置提交回调
        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setCommitCallback(new CommitCallback() {
            @Override
            public void onSuccess(Collection<TopicPartitionOffset> offsets) {
                // 偏移量提交成功，执行跟踪或后续操作
                offsets.forEach(offset -> 
                    System.out.println("偏移量提交成功：topic=" + offset.topic() + 
                        ", partition=" + offset.partition() + ", offset=" + offset.offset()));
            }

            @Override
            public void onFailure(Collection<TopicPartitionOffset> offsets, Exception exception) {
                // 偏移量提交失败，处理异常或记录日志
                System.err.println("偏移量提交失败：" + exception.getMessage());
            }
        });
        
        return factory;
    }
}

注意：该回调针对批量提交的偏移量触发（自动提交频率由spring.kafka.consumer.auto-commit-interval控制），适合需要跟踪偏移量最终提交状态的场景。

方案3：装饰器模式包装原有监听器（灵活定制处理逻辑） #

如果上述方案无法满足复杂需求，可使用装饰器模式包装现有的MessageListener或AcknowledgingMessageListener，在调用原监听器逻辑后执行自定义操作，同时保留自动提交偏移量特性。

• 定义装饰器类：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

public class PostProcessingMessageListenerDecorator<K, V> implements MessageListener<K, V> {

    private final MessageListener<K, V> delegate;

    public PostProcessingMessageListenerDecorator(MessageListener<K, V> delegate) {
        this.delegate = delegate;
    }

    @Override
    public void onMessage(ConsumerRecord<K, V> record) {
        // 调用原有监听器逻辑
        delegate.onMessage(record);
        // 消息处理后的自定义操作
        System.out.println("消息处理完成，偏移量：" + record.offset());
        // 执行偏移量跟踪或后续业务操作
    }
}

• 配置时替换原有监听器为装饰后的实例：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;

@Configuration
public class KafkaConfig {

    // 原有监听器Bean
    @Bean
    public MessageListener<?, ?> originalMessageListener() {
        // 此处为你原有的监听器实现，例如MessagingMessageListenerAdapter或自定义Listener
        return new MessagingMessageListenerAdapter(...);
    }

    // 包装后的监听器Bean
    @Bean
    public MessageListener<?, ?> decoratedMessageListener(MessageListener<?, ?> originalMessageListener) {
        return new PostProcessingMessageListenerDecorator<>(originalMessageListener);
    }

    // 后续容器配置使用decoratedMessageListener即可
}

该方案灵活性最高，可完全自定义消息处理前后的逻辑，且无需修改原有监听器代码。

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