Spring Integration Kafka主题名错误不抛异常的处理方案咨询

问题根因 #

你遇到的静默失败问题由两个核心配置原因导致：

• 你当前Kafka出站适配器配置了sync="false"，属于异步发送模式：发送动作只会把消息提交到Kafka Producer的本地缓冲区就直接返回，不会等待Kafka服务端的ACK响应。主题不存在的错误会在Producer的异步回调线程中抛出，不会传递到调用线程，因此你配置的requestHandlerAdvice、retryAdvice只能捕获调用线程的异常，根本感知不到异步回调里的发送失败。
• 若你的Kafka集群开启了auto.create.topics.enable=true（默认开启），写错的主题会被集群自动创建，也不会抛出未知主题异常。

解决方案 #

方案1：切换为同步发送（适合吞吐量要求不高的场景） #

直接修改出站适配器的sync属性为true：

<int-kafka:outbound-channel-adapter
                                        kafka-template="kafkaTemplate"
                                        auto-startup="true"
                                        topic="topicName"
                                        sync="true" >
        <int-kafka:request-handler-advice-chain>
            <ref bean="requestHandlerAdvice"/>
            <ref bean="retryAdvice"/>
        </int-kafka:request-handler-advice-chain>
</int-kafka:outbound-channel-adapter>

该模式下发送动作会阻塞等待Kafka服务端返回ACK，主题不存在、发送超时等错误会直接抛到调用线程，可直接被你现有的通知链捕获，你可以直接在requestHandlerAdvice中处理成功/失败逻辑并存入数据库。

方案2：保留异步发送，配置成功/失败通道（适合高吞吐量场景） #

不需要修改sync属性，通过Spring Integration Kafka提供的异步回调通道处理结果：

1. 给出站适配器新增成功、失败通道配置：

<int-kafka:outbound-channel-adapter
                                        kafka-template="kafkaTemplate"
                                        auto-startup="true"
                                        topic="topicName"
                                        sync="false"
                                        send-success-channel="kafkaSendSuccessChannel"
                                        send-failure-channel="kafkaSendFailureChannel" >
        <int-kafka:request-handler-advice-chain>
            <ref bean="retryAdvice"/>
        </int-kafka:request-handler-advice-chain>
</int-kafka:outbound-channel-adapter>

2. 分别编写两个通道的处理器，实现发送结果入库逻辑：

• 成功通道会收到携带发送结果的普通消息，可直接记录发送成功状态
• 失败通道会收到ErrorMessage，payload为KafkaSendFailureException，可从中提取原始消息、异常类型、错误主题等信息，记录发送失败状态

3. 补充Kafka Producer配置，确保错误能被正确感知：
   给DefaultKafkaProducerFactory的配置Map新增以下参数：

<entry key="acks" value="all"/>
<entry key="retries" value="3"/>
<entry key="delivery.timeout.ms" value="30000"/>
<!-- 若不需要自动创建主题，可在Kafka服务端将auto.create.topics.enable设为false，写错主题会直接返回UNKNOWN_TOPIC_OR_PARTITION异常 -->

注意事项 #

• 异步模式下不要依赖普通的请求处理通知做失败判断，必须通过send-failure-channel获取异步发送的异常
• 如果你需要在异步模式下实现重试，除了配置Producer的retries参数，也可以结合RetryTemplate + ErrorMessageSendingRecoverer实现自定义重试逻辑，重试耗尽后再将错误消息转发到失败通道入库。

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