使用低级Kafka Stream API进行context.forward()时的NPE [英] NPE while doing context.forward() using low-level Kafka Stream API

问题描述

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

I have built a plain Kafka streams API using Low-level Kafka API. The topology is linear.

p1->p2->p3

p1 -> p2 -> p3

在执行context.forward()时，我正在获取NPE，代码段如下:

While doing context.forward(), I am getting NPE, snippet here:

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.

I am using Kafka Stream 2.3.0.

我在[此处] [1]看到了一个类似的SO问题，该问题基于非常老的版本.因此，不确定这是否是相同的错误吗?

I see a similar SO question [here][1], and the question is based on the very old version. So, not sure if this is the same error?

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

I am putting some more info, keeping the Gist of what I am doing

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

推荐答案

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

It looks like it could be the same issue as the linked question, although we are talking about a much more contemporary version in your case. Make sure that ProcessorSupplier.get() returns a new instance each time it is called.

这篇关于使用低级Kafka Stream API进行context.forward()时的NPE的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！