Kafka Streams 默认不会在处理器抛出异常时提交对应输入记录，导致重启后重复处理；本文介绍通过 try/catch 主动捕获异常、结合流分支（split）将成功/失败记录分别路由至主输出与死信队列，从而实现语义可控的“至少一次”+DLQ保障。
kafka streams 默认不会在处理器抛出异常时提交对应输入记录，导致重启后重复处理；本文介绍通过 try/catch 主动捕获异常、结合流分支（split）将成功/失败记录分别路由至主输出与死信队列，从而实现语义可控的“至少一次”+dlq保障。
在 Kafka Streams 应用中，若 Processor（如 CustomProcessor）内部发生未捕获异常（例如 NullPointerException），框架会终止当前线程并触发 UncaughtExceptionHandler——但该异常发生时，对应输入记录尚未被提交（commit）。因此，当应用重启后，消费者将从上次已提交的 offset 处继续拉取，导致该异常记录被重复处理，形成无限重试循环。这不仅影响处理时效，还可能加剧下游故障。
根本解法不是依赖异常处理器“事后补救”，而是将异常处理逻辑前移至业务流程内，实现 record-level 的显式分流与状态标记。推荐采用以下模式：
✅ 推荐实践：基于 process() + split() 的可控异常路由
改写 CustomProcessor，不再让异常逃逸，而是统一捕获并输出带状态标识的结果：
KStream<String, String> messageStream = builder.stream(inputTopic);
// 使用 process() 显式处理，输出 (key, value, isValid) 三元组
KStream<String, KeyValue<String, Boolean>> enrichedStream = messageStream
.process(() -> new Processor<String, String, String, KeyValue<String, Boolean>>() {
private ProcessorContext context;
@Override
public void init(ProcessorContext context) {
this.context = context;
}
@Override
public void process(String key, String value) {
try {
String result = doBusinessLogic(value); // 你的核心处理逻辑
// 成功：输出有效结果 + 标记 true
context.forward(key, new KeyValue<>(result, true));
} catch (Exception e) {
// 失败：记录日志，输出原始输入 + 标记 false（便于 DLQ 消费端解析）
log.warn("Processing failed for key={}, value={}", key, value, e);
context.forward(key, new KeyValue<>(value, false));
}
}
private String doBusinessLogic(String value) {
// 示例：可能 NPE 的逻辑
return value.toUpperCase().trim(); // 若 value == null 则抛 NPE
}
});
// 将流按 isValid 字段拆分为两条子流
KStream<String, String>[] branches = enrichedStream
.split(Named.as("dlq-or-main"))
.branch((key, kv) -> !kv.value, Branched.withConsumer(ks -> ks.to(dlqTopic))) // 分支1：isValid == false → DLQ
.branch((key, kv) -> kv.value, Branched.withConsumer(ks -> ks.mapValues(kv -> kv.key).to(outputTopic))); // 分支2：isValid == true → 主输出? 关键点说明：
- process() 中 绝不让异常向上抛出，所有异常均被 catch 并转化为结构化输出；
- 使用 KeyValue
作为中间值类型，清晰区分成功/失败； - split() 配合 branch() 实现零拷贝逻辑分发，性能高效且语义明确；
- DLQ 中保留原始 value（或可扩展为包含异常堆栈、时间戳、traceId 的富对象），便于后续诊断与重放。
⚠️ 注意事项与最佳实践
- 禁用全局 UncaughtExceptionHandler 的自动恢复逻辑：若仍配置了 StreamsConfig.UNCAUGHT_EXCEPTION_HANDLER_CLASS_CLASS，请确保其仅做告警与监控，切勿调用 context.close() 或重启流，否则会干扰上述手动控制流。
- 避免在 process() 中执行阻塞 I/O 或长耗时操作：Kafka Streams 是单线程模型（每个 task 一个线程），阻塞会导致吞吐骤降甚至心跳超时。
- DLQ 消费端需幂等设计：因 DLQ 本身也基于 Kafka，其消费仍需考虑重复投递，建议在 DLQ 处理器中加入去重或幂等写入（如数据库 upsert）。
- 补充监控指标：通过 KafkaStreams#setMetricsRecordingLevel() 启用 INFO 级别指标，并监听 stream-task-created, stream-task-closed-revoked, record-lateness-max 等关键指标，及时发现异常高频段。
通过将异常处理内聚于流处理逻辑中，并借助 Kafka Streams 原生的 split 能力进行语义化分流，你不仅能彻底规避“重复处理-崩溃-再重复”的恶性循环，还能构建可观测、可运维、符合生产级 SLA 的流处理管道。
