KStream-KTable Join中Tombstone记录处理异常及解决办法

问题场景 #

用户搭建的KStream拓扑代码如下：

KStream<String, Data> eventStream = streamsBuilder.stream("event",
        Consumed.with(Serdes.String(), eventSerde));

KTable<String, Result> resultKTable = streamsBuilder.table("result",
        Consumed.with(Serdes.String(), resultSerde));

eventStream.leftJoin(resultKTable, new Joiner())
        .to("result", Produced.with(Serdes.String(), resultSerde));

向event主题发送Tombstone记录（值为null的记录）时，触发如下错误日志：

Skipping record due to null join key or value. key=[3428642] value=[null] topic=[event] partition=[0] offset=[21]

用户自行实现的临时处理方案：

eventStream.filter((k,v) -> v != null)
           .leftJoin(resultKTable, new Joiner())
           .to("result", Produced.with(Serdes.String(), resultSerde));
eventStream.filter((k,v) -> v == null).to("result");

问题原因 #

Kafka Streams的leftJoin操作默认会跳过值为null的输入记录（即Tombstone）。因为Join逻辑需要基于非空的流数据去关联表中的对应数据，当流中出现Tombstone时，框架会判定该记录无法参与Join，因此触发跳过日志且不执行Join逻辑。

方案解析与优化 #

你的临时方案拆分了两种记录的处理逻辑：非空记录走Join流程，Tombstone直接转发到结果主题，这个思路完全符合业务语义。可以通过split()+branch()优化代码结构，避免重复过滤同一个流：

eventStream.split()
    .branch((k, v) -> v != null, Branched.withConsumer(stream -> 
        stream.leftJoin(resultKTable, new Joiner())
              .to("result", Produced.with(Serdes.String(), resultSerde))
    ))
    .branch((k, v) -> v == null, Branched.withConsumer(stream -> 
        stream.to("result")
    ));

另外需要注意：将Tombstone写入result主题后，由于该主题同时被resultKTable消费，会触发KTable中对应key记录的删除操作，这完全符合Tombstone的语义（标记删除对应key的状态），如果这是你的业务预期，那么该方案是合理且正确的。

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