如何在Kafka Stream拓扑中跳过抛出运行时错误的消息

我用常规方式创建了Kafka Stream拓扑，代码如下：

public Topology buildTopology() {
    StreamsBuilder builder = new StreamsBuilder();

    builder.stream(inputTopic, Consumed.with(...))
        .filter(this::filterEvents)
        .mapValues(this::mapToOtherEvent)
        .to(outputTopic, Produced.with(...));

    return builder.build();
}

其中mapToOtherEvent方法会偶尔抛出业务逻辑异常，导致整个处理流程中断。我尝试过以下方案但都不符合需求：

• 实现StreamsUncaughtExceptionHandler：仅能处理异常并重放消息，陷入循环；
• 了解ProductionExceptionHandler和DeserializationExceptionHandler：二者不处理业务逻辑异常；
• 参考Stack Overflow上的代码方案：在我的场景中会产生其他影响。

可行解决方案：在业务处理层捕获异常并跳过消息 #

核心思路是在业务逻辑调用时主动捕获异常，避免异常扩散到Kafka Streams框架，同时可选将错误消息转发到死信队列（DLQ）用于后续排查。

修改后的代码示例：

public Topology buildTopology() {
    StreamsBuilder builder = new StreamsBuilder();

    // 可选：定义死信队列主题
    String dlqTopic = inputTopic + "-dlq";

    builder.stream(inputTopic, Consumed.with(...))
        .filter(this::filterEvents)
        .flatMapValues((key, value) -> {
            try {
                // 调用业务方法，处理正常消息
                return Collections.singletonList(mapToOtherEvent(value));
            } catch (BusinessException e) {
                // 记录异常日志
                log.error("处理消息失败，key: {}, value: {}, 异常: {}", key, value, e.getMessage(), e);
                // 将错误消息发送到死信队列
                producer.send(new ProducerRecord<>(dlqTopic, key, value));
                // 返回空列表，跳过当前消息
                return Collections.emptyList();
            }
        })
        .to(outputTopic, Produced.with(...));

    return builder.build();
}

方案说明 #

1. 用flatMapValues替代mapValues：flatMap支持返回空列表来跳过当前消息，而mapValues必须返回非null值；
2. 主动捕获业务异常：在try-catch块中包裹业务方法调用，防止异常触发Kafka Streams的全局异常处理逻辑；
3. 死信队列（可选）：将错误消息转发到DLQ，保留异常现场，方便后续分析或重试；
4. 流程不中断：捕获异常后返回空列表，当前消息会被跳过，后续消息正常处理。

注意事项 #

• 如果不需要死信队列，可去掉发送DLQ的逻辑，直接返回空列表；
• 建议捕获明确的业务异常类型（比如自定义的BusinessException），避免捕获所有Exception导致隐藏其他问题；
• 日志中需记录足够的信息（消息key、value、异常栈），便于定位问题。

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