Kafka Stream count on time window not reporting zero values

Viewed 1732

I'm using a Kafka streams to calculate how many events occurred in last 3 minutes using a hopping time window:

public class ViewCountAggregator {

    void buildStream(KStreamBuilder builder) {      

        final Serde<String> stringSerde = Serdes.String();
        final Serde<Long> longSerde = Serdes.Long();

        KStream<String, String> views = builder.stream(stringSerde, stringSerde, "streams-view-count-input");
        KStream<String, Long> viewCount = views
            .groupBy((key, value) -> value)
            .count(TimeWindows.of(TimeUnit.MINUTES.toMillis(3)).advanceBy(TimeUnit.MINUTES.toMillis(1)))
            .toStream()
            .map((key, value) -> new KeyValue<>(key.key(), value));

        viewCount.to(stringSerde, longSerde, "streams-view-count-output");        
    }

    public static void main(String[] args) throws Exception {                   
        // some not so important initialization code
        ...  
    }

}

When running a consumer and pushing some messages to an input topic it receives following updates as the time passes:

single  1
single  1
single  1
five    1
five    4
five    5
five    4
five    1

Which is almost correct, but it never receives updates for:

single  0
five    0

Without it my consumer that updates a counter will never set it back to zero when there are no events for a longer period of time. I'm expecting consumed messages to look like this:

single  1
single  1
single  1
single  0
five    1
five    4
five    5
five    4
five    1
five    0

Is there some configuration option / argument I'm missing that would help me achieving such behavior?

1 Answers
Related