diff --git a/kafka-streams-samples/kafka-streams-word-count/README.adoc b/kafka-streams-samples/kafka-streams-word-count/README.adoc index 82cbc94..e6e1e4e 100644 --- a/kafka-streams-samples/kafka-streams-word-count/README.adoc +++ b/kafka-streams-samples/kafka-streams-word-count/README.adoc @@ -6,6 +6,7 @@ The example is based on the word count application from the https://github.com/c It uses a single input and a single output. In essence, the application receives text messages from an input topic and computes word occurrence counts in a configurable time window and report that in an output topic. This sample uses lambda expressions and thus requires Java 8+. +The sample uses a default timewindow of 30 seconds. === Running the app: @@ -15,7 +16,7 @@ Go to the root of the repository. `./mvnw clean package` -`java -jar target/kafka-streams-word-count-0.0.1-SNAPSHOT.jar --spring.cloud.stream.kafka.streams.timeWindow.length=60000` +`java -jar target/kafka-streams-word-count-0.0.1-SNAPSHOT.jar` Assuming you are running the dockerized Kafka cluster as above. @@ -29,13 +30,4 @@ On another terminal: Enter some text in the console producer and watch the output in the console consumer. -Time window can be changed using the following property. - -`spring.cloud.stream.kafka.streams.timeWindow.length` (value is expressed in milliseconds) - -In order to switch to a hopping window, you can use the `spring.cloud.stream.kstream.timeWindow.advnceBy` (value in milliseconds). -This will create an overlapped hopping windows depending on the value you provide. - -Here is an example with 2 overlapping windows (window length of 10 seconds and a hop (advance) by 5 seconds: - -`java -jar target/kafka-streams-word-count-0.0.1-SNAPSHOT.jar --spring.cloud.stream.kafka.streams.timeWindow.length=10000 --spring.cloud.stream.kafka.streams.timeWindow.advnceBy=5000` \ No newline at end of file +Once you are done, stop the Kafka cluster: `docker-compose down` \ No newline at end of file diff --git a/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java b/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java index b18f201..7e6fcf9 100644 --- a/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java +++ b/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java @@ -22,7 +22,6 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.Serialized; import org.apache.kafka.streams.kstream.TimeWindows; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.stream.annotation.EnableBinding; @@ -43,9 +42,6 @@ public class KafkaStreamsWordCountApplication { @EnableBinding(KafkaStreamsProcessor.class) public static class WordCountProcessorApplication { - @Autowired - private TimeWindows timeWindows; - @StreamListener("input") @SendTo("output") public KStream process(KStream input) { @@ -54,7 +50,7 @@ public class KafkaStreamsWordCountApplication { .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) - .windowedBy(timeWindows) + .windowedBy(TimeWindows.of(30000)) .count(Materialized.as("WordCounts-1")) .toStream() .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));