如何配置int-kafka消息驱动通道适配器的恢复回退、错误处理器及手动提交偏移？

一、错误处理器与恢复回退配置 #

1. 配置Spring Kafka错误处理器（带恢复逻辑） #

定义SeekToCurrentErrorHandler配合RecoveryCallback实现重试耗尽后的恢复逻辑，并关联到你的ConcurrentMessageListenerContainer：

<!-- 恢复回调：处理重试耗尽后的逻辑 -->
<bean id="kafkaRecoveryCallback" class="org.springframework.kafka.listener.RecoveryCallback<Void>">
    <lambda>
        <![CDATA[
            (context) -> {
                ConsumerRecord<?, ?> failedRecord = context.getLastThrowable().getFailedRecord();
                // 自定义恢复逻辑：如记录错误日志、投递死信队列等
                System.err.println("重试耗尽，失败消息：topic=" + failedRecord.topic() + ", offset=" + failedRecord.offset());
                return null;
            }
        ]]>
    </lambda>
</bean>

<!-- 错误处理器：设置重试次数与恢复回调 -->
<bean id="kafkaErrorHandler" class="org.springframework.kafka.listener.SeekToCurrentErrorHandler">
    <constructor-arg ref="kafkaRecoveryCallback"/>
    <constructor-arg value="3"/> <!-- 最大重试次数 -->
</bean>

<!-- 修改listener container，关联错误处理器 -->
<bean id="testEventListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer">
    <constructor-arg ref="consumerFactoryTestEvent"/>
    <property name="concurrency" value="3"/>
    <constructor-arg>
        <bean class="org.springframework.kafka.listener.ContainerProperties">
            <constructor-arg name="topics" value="xyz"/>
            <property name="ackMode" value="MANUAL_IMMEDIATE"/>
            <property name="errorHandler" ref="kafkaErrorHandler"/> <!-- 绑定错误处理器 -->
        </bean>
    </constructor-arg>
</bean>

2. 结合Spring Integration Error Channel处理 #

若需要更灵活的错误分流，可利用配置的errorChannel实现自定义错误处理：

<!-- 错误通道消息处理器 -->
<int:service-activator input-channel="errorChannel" ref="errorHandlerService" method="handleError"/>

<bean id="errorHandlerService" class="com.yourpackage.KafkaErrorHandlerService"/>

对应的Java实现类：

public class KafkaErrorHandlerService {
    public void handleError(ErrorMessage errorMessage) {
        Message<?> originalMsg = errorMessage.getOriginalMessage();
        // 获取原始Kafka记录与异常信息
        ConsumerRecord<?, ?> failedRecord = originalMsg.getHeaders().get(KafkaHeaders.RECORD, ConsumerRecord.class);
        Throwable exception = errorMessage.getPayload();
        
        // 自定义错误处理逻辑
        System.err.println("处理错误：topic=" + failedRecord.topic() + ", error=" + exception.getMessage());
        
        // 按需手动提交偏移量（根据业务决定是否确认失败消息）
        Acknowledgment ack = originalMsg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
        if (ack != null) {
            ack.acknowledge();
        }
    }
}

二、自定义对象场景下的手动提交偏移量 #

当处理自定义对象时，Spring Integration会将Kafka消息封装为Message，其中KafkaHeaders.ACKNOWLEDGMENT头携带了提交偏移量所需的Acknowledgment对象，可通过以下两种方式获取：

1. 服务方法通过注解获取Header参数 #

在消息处理Bean的方法中，用@Header直接注入Acknowledgment：

@Service
public class KafkaMessageHandler {
    public void handleCustomMessage(@Payload YourCustomObject payload,
                                   @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
        try {
            // 处理自定义对象业务逻辑
            processCustomData(payload);
            // 处理完成后手动提交偏移量
            acknowledgment.acknowledge();
        } catch (Exception e) {
            // 异常抛出后，错误处理器会触发重试/恢复逻辑
            throw new RuntimeException("消息处理失败", e);
        }
    }
    
    private void processCustomData(YourCustomObject payload) {
        // 自定义业务实现
    }
}

对应的Spring Integration配置：

<int:service-activator input-channel="kafka-input-channel-test-Event"
                       ref="kafkaMessageHandler"
                       method="handleCustomMessage"/>

2. 从MessageHeaders中手动提取 #

若直接操作Message对象，可从Header中获取Acknowledgment：

public void handleMessage(Message<YourCustomObject> message) {
    YourCustomObject payload = message.getPayload();
    Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
    
    try {
        processCustomData(payload);
        acknowledgment.acknowledge();
    } catch (Exception e) {
        throw new RuntimeException("处理失败", e);
    }
}

关键注意事项 #

• 使用MANUAL_IMMEDIATE ack模式时，必须手动调用acknowledge()，否则偏移量不会提交，消息会被重复消费。
• 恢复逻辑需结合业务场景设计，比如将失败消息投递到死信队列，避免无限重试占用资源。
• 重试次数需根据业务容忍度配置，避免过度重试导致系统负载过高。

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