Hello Tim, I think I found the issue, could you apply the following patch and retry your app?
https://github.com/apache/kafka/pull/1520 Guozhang On Thu, Jun 16, 2016 at 11:11 PM, Guozhang Wang <wangg...@gmail.com> wrote: > Hello Tim, > > By looking through the source code I suspect it is a bug in Kafka Stream's > KStreamWindowReduce > implementation. I'll do further investigation tomorrow and possibly file a > JIRA with a patch. > > > Guozhang > > > On Thu, Jun 16, 2016 at 4:12 PM, Tim Renner <renner.timo...@gmail.com> > wrote: > >> Hi all, >> >> I'm trying to do a KTable-KTable join to compute an average within a >> tumbling window. >> Here's the KStreams code (I've put a fully working example in a gist: >> https://gist.github.com/timothyrenner/a99c86b2d6ed2c22c8703e8c7760af3a) >> KStreamBuilder builder = new KStreamBuilder(); >> >> KStream<String, Long> longs = builder.stream( >> Serdes.String(), Serdes.Long(), "longs"); >> >> KTable<Windowed<String>, Long> longCounts = >> longs.countByKey(TimeWindows.of("longCounts", 10000L), >> Serdes.String()); >> >> >> KTable<Windowed<String>, Long> longSums = >> longs.reduceByKey((v1, v2) -> v1 + v2, >> TimeWindows.of("longSums", 10000L), >> Serdes.String(), >> Serdes.Long()); >> >> KTable<Windowed<String>, Double> longAvgs = >> longSums.join(longCounts, >> (sum, count) -> >> sum.doubleValue()/count.doubleValue()); >> >> longAvgs.toStream((wk, v) -> wk.key()) >> .to(Serdes.String(), >> Serdes.Double(), >> "long-avgs"); >> >> KafkaStreams streams = new KafkaStreams(builder, config); >> streams.start(); >> >> When I run this, I get the following exception: >> >> java.util.NoSuchElementException >> at >> >> org.apache.kafka.streams.state.internals.RocksDBWindowStore$RocksDBWindowStoreIterator.next(RocksDBWindowStore.java:95) >> at >> >> org.apache.kafka.streams.state.internals.RocksDBWindowStore$RocksDBWindowStoreIterator.next(RocksDBWindowStore.java:64) >> at >> >> org.apache.kafka.streams.state.internals.MeteredWindowStore$MeteredWindowStoreIterator.next(MeteredWindowStore.java:136) >> at >> >> org.apache.kafka.streams.state.internals.MeteredWindowStore$MeteredWindowStoreIterator.next(MeteredWindowStore.java:117) >> at >> >> org.apache.kafka.streams.kstream.internals.KStreamWindowReduce$KStreamAggregateValueGetter.get(KStreamWindowReduce.java:166) >> at >> >> org.apache.kafka.streams.kstream.internals.KStreamWindowReduce$KStreamAggregateValueGetter.get(KStreamWindowReduce.java:147) >> at >> >> org.apache.kafka.streams.kstream.internals.KTableKTableJoin$KTableKTableJoinProcessor.process(KTableKTableJoin.java:77) >> at >> >> org.apache.kafka.streams.kstream.internals.KTableKTableJoin$KTableKTableJoinProcessor.process(KTableKTableJoin.java:48) >> at >> >> org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:68) >> at >> >> org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:338) >> at >> >> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187) >> at >> >> org.apache.kafka.streams.kstream.internals.KStreamWindowAggregate$KStreamWindowAggregateProcessor.process(KStreamWindowAggregate.java:136) >> at >> >> org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:68) >> at >> >> org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:338) >> at >> >> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187) >> at >> >> org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:64) >> at >> >> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:174) >> at >> >> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:320) >> at >> >> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218) >> >> Looks like the join is throwing the exception. Any ideas? >> >> Thanks, >> Tim >> > > > > -- > -- Guozhang > -- -- Guozhang