使用Kafka Streams 3.3.1构建拓扑时遭遇NullPointerException求助

问题原因 #

这个空指针异常是Kafka Streams 3.3.1版本的已知bug。当你在selectKey()之后直接调用groupByKey()时，内部会生成一个重分区节点（RepartitionNode），但该版本的ProcessorParameters.toString()方法会尝试调用未初始化的processorSupplier.get()，从而抛出NPE。从异常栈的调用链能看出，问题出在toString方法的执行中，而非拓扑实际运行的业务逻辑。

解决方案 #

有两种可行的解决方式：

1. 升级Kafka Streams版本 #

该bug在3.3.2及更高版本中已被修复，直接将依赖升级到3.3.2或最新稳定版即可解决问题。

2. 修改代码逻辑（不升级版本的情况下） #

去掉selectKey()步骤，直接用groupBy()替代selectKey()+groupByKey()的组合，避免触发有问题的重分区节点逻辑。修改后的代码如下：

private static Topology createTopology() {
    StreamsBuilder builder = new StreamsBuilder();
    KTable<Integer, Long> table = builder.stream("messages")
            // 直接通过groupBy指定分组key，替代selectKey+groupByKey的组合
            .groupBy((key, value) -> 1)
            .count();
    table.toStream().to("stats");
    return builder.build();
}

调整后，拓扑构建时不会生成那个存在问题的重分区节点，也就不会触发空指针异常了。

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