使用低级 Kafka Stream API 执行 context.forward() 时的 NPE答案

【问题标题】：NPE while doing context.forward() using low-level Kafka Stream API使用低级 Kafka Stream API 执行 context.forward() 时的 NPE
【发布时间】：2023-04-08 20:36:01
【问题描述】：

我使用低级 Kafka API 构建了一个普通的 Kafka 流 API。拓扑是线性的。

p1 -> p2 -> p3

在执行 context.forward() 时，我在这里得到 NPE，sn-p：

NAjava.lang.NullPointerException: null
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:178)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)

...

我正在使用 Kafka Stream 2.3.0。

我在 [此处][1] 看到了一个类似的 SO 问题，该问题基于非常旧的版本。那么，不确定这是否是同一个错误？

编辑

我正在提供更多信息，保持我正在做的事情的要点

public class SP1Processor implements StreamProcessor {

private StreamProcessingContext ctxt;

// In init(), create a single thread pool
// which does some processing and sends the
// data to next processor
@Override
void init(StreamProcessingContext ctxt) {

      this.ctxt = ctxt;

     // Create a thread pool, do some work
     // and then do this.ctxt.forward(K,V)

    // Not showing code of Thread pool
    // Strangely, inside this thread pool,
    // this.ctxt isn't same what I see in process()
    // shouldn't it be same? ctxt is member variable
    // and shouldn't it be same
    // this.ctxt.forward(K,V) here in this thread pool is causing NPE
    // why does it happen?
    this.ctxt.forward(K,V);

}

@Override
void process(K,V) {

   // Here do some processing and go to the next processor chain
   // This works fine
   this.ctxt.forward(K,V);
}

}

  [1]: https://stackoverflow.com/questions/39067846/periodic-npe-in-kafka-streams-processor-context

【问题讨论】：

你能添加代码sinpped吗？
@BartoszWardziński：我添加了代码 sn-p。在 process() 中， ctxt.forward() 效果很好；但是，在 init() 中，我生成了一个线程池，并且当我执行 ctxt.forward() 时，它会导致 NPE。 ctxt 本身在 process() 或线程池中都不为 null，但奇怪的是它们不一样

标签： java apache-kafka apache-kafka-streams

【解决方案1】：

看起来它可能与链接的问题是同一个问题，尽管在您的情况下我们正在谈论一个更现代的版本。确保 ProcessorSupplier.get() 在每次调用时都返回一个新实例。

【讨论】：

ProcessorSupplier.get() 是仅在拓扑构建期间调用还是在我们执行 context.forward() 时调用？
我加了代码sn-p，请大家看看有什么不寻常的地方告诉我？

【解决方案2】：

您不应在处理器或 DSL 调用中创建任何线程池。 KafkaStreams 中的并行性由num.stream.threads、分区数和实例数管理。

ctxt 相同，但其字段/成员可能不同（例如currentNode） - 可能会被不同的线程更改。

【讨论】：