カフカストリームプロセッサコンテキストでの定期NPE

Question

kafka-streams 0.10.0.0を使用して、メッセージを転送するときにStreamTaskでヌルポインタ例外が定期的に発生しています。呼び出しの10％から50％の間で変化します。 NPEは、この方法で行われます。

public <K, V> void forward(K key, V value) { 
    ProcessorNode thisNode = currNode; 
    try { 
     for (ProcessorNode childNode : (List<ProcessorNode<K, V>>) thisNode.children()) { 
      currNode = childNode; 
      childNode.process(key, value); 
     } 
    } finally { 
     currNode = thisNode; 
    } 
} 

いくつかのケースでは、thisNodeフィールドがヌルであると思われます。これを引き起こしているのは何ですか？スタックトレースは以下の通りです。

[ERROR] 2016-08-21 14:50:39.288 [StreamThread-1] StreamedMetricMeter - Forwarding failed 
java.lang.NullPointerException 
    at org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:336) ~[kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187) ~[kafka-streams-0.10.0.0.jar:?] 
    at com.heliosapm.streams.metrics.processors.AbstractStreamedMetricProcessor.forward(AbstractStreamedMetricProcessor.java:552) [classes/:?] 
    at com.heliosapm.streams.metrics.processors.impl.StreamedMetricMeter.doProcess(StreamedMetricMeter.java:89) [classes/:?] 
    at com.heliosapm.streams.metrics.processors.impl.StreamedMetricMeter.doProcess(StreamedMetricMeter.java:1) [classes/:?] 
    at com.heliosapm.streams.metrics.processors.AbstractStreamedMetricProcessor.process(AbstractStreamedMetricProcessor.java:166) [classes/:?] 
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:68) [kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:338) [kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187) [kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:64) [kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:174) [kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:320) [kafka-streams-0.10.0.0.jar:?] 
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218) [kafka-streams-0.10.0.0.jar:?] 

Answer 1

問題は私のProcessorSupplier sがgetへのすべての呼び出しのためにプロセッサの同じインスタンスを返したということでした。次に、Kafka Streamsエンジンは複数のプロセッサインスタンスを作成しようとしていましたが、疑いなくマルチスレッドダンプスター火災が発生しました。同じように注意しないでください.... ProcessorSupplier.get（）は、各呼び出しでプロセッサの新しいインスタンスを返す必要があります。