Kafka Streams遇转换异常时如何跳过异常记录并继续处理？

场景与需求 #

你的Kafka Streams应用流程大致如下：

final StreamsBuilder builder = new StreamsBuilder();

final KStream<String, String> textLines = builder.stream(inputTopic);

final KStream<String, String> textTransformation_1 = textLines.processValues(value ->  value+"firstTranstormation");  

final KStream<String, String> textTransformation_2 = textTransformation_1.processValues(value ->  value+"secondTranstormation"); 

// 核心关注步骤
final KStream<String, String> textTransformation_3 = textTransformation_2.processValues(this::processValueAndDoRelatedStuff); 

// ...后续处理步骤
textTransformation_x.to(outputTopic, Produced.with(Serdes.String(), Serdes.Long()));

final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
streams.start();
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

你希望：当processValueAndDoRelatedStuff(String input)抛出错误时，不终止整个程序，仅忽略该条记录的转换结果（不发送到outputTopic），同时正常处理其他消息。

你已经想到一种方案：在processValueAndDoRelatedStuff中捕获异常并返回特定标记值，后续通过filter过滤掉该值：

final KStream<String, String> textTransformation_4 = textTransformation_3.filter((k,v) -> !v.equals("badrecord")); 

但你更关心另一种场景：如果映射函数未捕获异常直接抛出，Kafka Streams能否自动忽略这条引发异常的记录，继续处理其余消息？

解答 #

1. 默认行为：程序会崩溃（或反复重启） #

默认情况下，Kafka Streams遇到处理阶段未捕获的异常时，会判定为不可恢复的错误，直接终止当前流处理任务。如果你的应用配置了重启策略（比如默认的无限重试重启），程序会反复崩溃重启，但永远卡在引发异常的那条记录上，不会自动跳过它。

2. 需求可行，但必须主动处理异常 #

要实现“忽略异常记录、继续处理其他消息”的需求，不能依赖Kafka Streams的默认行为，必须通过业务逻辑层面的处理来实现，以下是两种更优雅的方式：

方式一：用flatMapValues替代processValues，在内部处理异常

flatMapValues允许你返回一个结果集合——正常处理时返回包含转换后值的单元素集合，异常时返回空集合，这样这条记录会被自动忽略，无需额外的filter步骤：

final KStream<String, String> textTransformation_3 = textTransformation_2.flatMapValues(value -> {
    try {
        return Collections.singletonList(processValueAndDoRelatedStuff(value));
    } catch (Exception e) {
        // 这里可以记录异常日志，方便排查问题
        log.error("处理记录失败，将忽略该条记录", e);
        return Collections.emptyList();
    }
});

方式二：自定义ValueTransformer处理异常

如果你的处理逻辑更复杂，可以自定义ValueTransformerWithKey，在transform方法中捕获异常并返回null，后续用filter过滤掉null值：

final KStream<String, String> textTransformation_3 = textTransformation_2.transformValues(() -> new ValueTransformerWithKey<String, String, String>() {
    @Override
    public void init(ProcessorContext context) {
        // 初始化逻辑（比如获取上下文）
    }

    @Override
    public String transform(String key, String value) {
        try {
            return processValueAndDoRelatedStuff(value);
        } catch (Exception e) {
            log.error("处理记录[key={}]失败", key, e);
            return null;
        }
    }

    @Override
    public void close() {
        // 资源清理逻辑
    }
});

// 过滤掉处理失败返回的null值
final KStream<String, String> filteredStream = textTransformation_3.filter((k, v) -> v != null);

3. 为什么不能依赖Kafka Streams自动跳过？ #

Kafka Streams的设计目标是保证数据处理的精确一次或至少一次语义，自动跳过异常记录会破坏这种语义一致性——它无法判断异常是临时的（比如网络波动）还是永久的（比如数据格式完全错误）。因此，框架把是否忽略的决定权交给了开发者，让你根据业务场景选择合适的处理方式。

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