Spring Cloud Kafka Streams异常处理：如何将错误消息发送到DLQ且不中断流处理

核心思路是在Function处理器内做单条消息的异常捕获，通过流分支将正常消息和异常消息拆分，异常消息直接投递到DLQ主题，正常消息走后续处理逻辑，从根源上避免异常抛到Kafka Streams框架层导致整个流停止。

第一步：定义处理结果封装类（可选但推荐） #

用来统一承载原始消息、处理结果、异常信息，方便后续分支判断：

@Data
@AllArgsConstructor
public class ProcessResult<T> {
    // 原始输入消息
    private String originalMessage;
    // 处理成功的返回结果
    private T data;
    // 处理失败的异常信息
    private Exception error;

    public boolean isSuccess() {
        return error == null;
    }
}

第二步：改造Function处理器，单条捕获异常 #

处理每条消息时包裹try-catch，不要让异常透出到框架层：

@Bean
public Function<KStream<String, String>, KStream<String, CityProgrammes>> cityProgrammesProcessor() {
    return input -> {
        // 1. 处理每条消息，捕获所有业务异常封装为结果对象
        KStream<String, ProcessResult<CityProgrammes>> processedStream = input.mapValues(rawStr -> {
            try {
                // 原有业务逻辑：调用API查城市信息、查周边活动
                CityInfo city = cityApi.queryByName(rawStr);
                CityProgrammes programmes = activityApi.queryByLocation(city.getLocation());
                return new ProcessResult<>(rawStr, programmes, null);
            } catch (Exception e) {
                // 所有异常全部捕获，封装为错误结果
                return new ProcessResult<>(rawStr, null, e);
            }
        });

        // 2. 分流：拆分成功流和失败流
        Map<String, KStream<String, ProcessResult<CityProgrammes>>> branch = processedStream
                .split(Named.as("process-"))
                .branch((k, v) -> v.isSuccess(), Branched.as("success"))
                .branch((k, v) -> !v.isSuccess(), Branched.as("fail"))
                .noDefaultBranch();

        // 3. 失败流直接投递到DLQ主题，可按需把异常信息拼接进消息体
        branch.get("process-fail")
                .mapValues(v -> String.format("原始消息:%s, 异常信息:%s", v.getOriginalMessage(), v.getError().getMessage()))
                .to("dlq-city-programmes-topic");

        // 4. 返回成功流，走后续正常处理逻辑
        return branch.get("process-success")
                .mapValues(ProcessResult::getData);
    };
}

第三步：可选配置优化 #

如果不想硬编码DLQ主题名，可以通过@Value注入配置文件中的DLQ主题值；如果需要自定义DLQ消息的序列化、重试策略，直接在配置文件中添加对应Kafka生产者属性即可。

补充说明 #

• 你之前调研的UncaughtExceptionHandler是Kafka Streams的全局异常处理器，确实无法获取具体出错消息，仅能做流重启等全局操作，完全不适合单条业务异常的处理场景。
• 该方案不需要修改下游被调用服务的异常处理逻辑，所有兜底逻辑都收敛在流处理层，符合架构边界要求。
• 只要try-catch覆盖了所有业务处理逻辑，不会出现消息丢失情况，异常消息全部进入DLQ，正常消息也不会被阻塞。

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