Spring Cloud Stream与Kafka函数式模型中KStream异常时如何发送至错误主题？

核心实现思路 #

放弃在catch块中直接手动发送消息的写法，基于Kafka Streams原生的流分支路由能力，结合Spring Cloud Stream的多输出绑定机制，实现成功/异常消息的自动分流，保证流处理的一致性。

具体实现步骤 #

1. 统一封装处理结果结构 #

定义通用的处理结果POJO，统一承载成功、异常场景的消息元数据，避免不同场景下的返回类型混乱：

// 可根据实际业务调整字段
public record ProcessResult(
    boolean processSuccess,
    String messageKey,
    Object rawMessageValue,
    String errorDetail,
    Object businessProcessedValue
) {}

2. 改造流处理函数拓扑 #

将原来单输出的流处理函数，调整为多输出结构，在业务处理的catch块中仅封装异常状态，不直接执行发消息操作，最后通过branch算子拆分出成功流、异常流：

public Function<KStream<String, ?>, KStream<String, ProcessResult>[]> process(){
    return input -> input
        .map((key, value) -> {
            try {
                // 原有业务处理逻辑
                Object processResult = customBusinessHandle(value);
                // 封装成功结果
                return new KeyValue<>(key, 
                    new ProcessResult(true, key, value, null, processResult)
                );
            } catch (Exception e) {
                // 仅记录异常信息封装到结果对象，不直接发消息
                return new KeyValue<>(key,
                    new ProcessResult(false, key, value, e.getMessage(), null)
                );
            }
        })
        // 拆分两条独立流
        .branch(
            // 第一个分支：处理成功的消息，对应第一个输出绑定
            (key, result) -> result.processSuccess(),
            // 第二个分支：处理失败的消息，对应第二个输出绑定
            (key, result) -> !result.processSuccess()
        );
}

3. 配置绑定规则 #

在应用配置文件中，将拆分出的两条流分别绑定到指定的物理主题，实现路由目标的自定义：

spring:
  cloud:
    stream:
      function:
        # 声明加载的函数定义
        definition: process
      bindings:
        # 原始消息输入主题
        process-in-0:
          destination: source-input-topic
        # 成功消息输出主题（对应branch返回的第一个流）
        process-out-0:
          destination: business-success-topic
        # 错误消息输出主题（对应branch返回的第二个流）
        process-out-1:
          destination: custom-error-topic
      kafka:
        streams:
          binder:
            # 配置流处理应用唯一ID
            applicationId: process-stream-app

关键注意事项 #

• 不要在catch块中直接注入KafkaTemplate手动发送消息：这种写法会脱离Kafka Streams的位点管控机制，遇到流任务重平衡、消费重试时会出现重复发消息、消息丢失的问题，还需要手动对齐序列化、分区规则，维护成本极高。
• 基于branch的路由方式是流处理拓扑的一部分，完全和流处理的位点进度对齐，不会出现一致性问题，还可以根据异常类型、消息标签继续拆分更多分支，实现更细粒度的主题路由。
• 对于未被业务代码catch的反序列化异常、运行时异常，可以搭配Spring Cloud Stream的默认DLQ机制兜底，配置spring.cloud.stream.kafka.streams.binder.deserialization-exception-handler: sendToDlq即可将未捕获异常的消息自动投递到绑定的死信主题。

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