Good morning,
I have a problem with a kafka-stream application.
In fact I 've created already two kafka stream applications :
StatsByMinute : entry topic : uplinks, out topic : statsM.
StatsByHour : entrey topic : statsM, out topic : statsH.
StatsByDay : entry topic : statsH, out topic : statsD.
The three of these application hava naerly the same struture ( StatsByMinute
and StatsBy Hour/Stats By Day are only different in the application ID KVstore
and the mthos process() ).
StatsBy Day and Stats BY Hour have exactly the same structure (the only
exception is the ID parameters) .
The Problem is that stastByMinute and StatsByHour works parfectly.
But this this not the case for StatsByDay where i verified that i do receive
data and process it (so process works). but in the line context.forward in
punctuate there is a problem.
I get the following error :
[2016-09-16 15:44:24,467] ERROR Streams application error during processing in
thread [StreamThread-1]:
(org.apache.kafka.streams.processor.internals.StreamThread)
java.lang.NullPointerException
at
org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:336)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187)
at com.actility.tpk.stat.BaseProcessor.punctuate(BaseProcessor.java:54)
at
org.apache.kafka.streams.processor.internals.StreamTask.punctuate(StreamTask.java:227)
at
org.apache.kafka.streams.processor.internals.PunctuationQueue.mayPunctuate(PunctuationQueue.java:45)
at
org.apache.kafka.streams.processor.internals.StreamTask.maybePunctuate(StreamTask.java:212)
at
org.apache.kafka.streams.processor.internals.StreamThread.maybePunctuate(StreamThread.java:407)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:325)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218)
Exception in thread "StreamThread-1" java.lang.NullPointerException
at
org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:336)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187)
at com.actility.tpk.stat.BaseProcessor.punctuate(BaseProcessor.java:54)
at
org.apache.kafka.streams.processor.internals.StreamTask.punctuate(StreamTask.java:227)
at
org.apache.kafka.streams.processor.internals.PunctuationQueue.mayPunctuate(PunctuationQueue.java:45)
at
org.apache.kafka.streams.processor.internals.StreamTask.maybePunctuate(StreamTask.java:212)
at
org.apache.kafka.streams.processor.internals.StreamThread.maybePunctuate(StreamThread.java:407)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:325)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218)
