By default, .windowedBy(SessionWindows.with(...)) will return every new incoming record. So, how could I wait for atleast 1 second before returning the last result of current session window?
I'm trying with word count example:
final KStream<String, String> source = builder.stream("streams-plaintext-input");
final KStream<String, Long> wordCounts = source
// Split each text line, by whitespace, into words.
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" ")))
// Group the stream by word to ensure the key of the record is the word.
.groupBy((key, word) -> word)
.windowedBy(SessionWindows.with(Duration.ofSeconds(10)))
// Count the occurrences of each word (message key).
.count(Materialized.with(Serdes.String(), Serdes.Long()))
.suppress(Suppressed.untilTimeLimit(Duration.ofSeconds(1), Suppressed.BufferConfig.unbounded()))
// Convert to KStream<String, Long>
.toStream((windowedId, count) -> windowedId.key());
wordCounts.foreach((word, count) -> {
System.out.println(word + " : " + count);
});
This is the input of producer and result in client, which is actually wrong:
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-plaintext-input
>hello kafka stream
(nothing)
>hello kafka stream
hello : 1
kafka : 1
stream : 1
>hello kafka stream
hello : null
kafka : 1
stream : 1
How could I fix that? Thanks a lot for reading my question :)