== What is this app?
This is an example of a Spring Cloud Stream processor using Kafka Streams support.
The example is based on a contrived use case of tracking products.
Although contrived, this type of use cases are pretty common in the industry where some organizations want to track the statistics of some entities over a time window.
In essence, the application receives product information from input topic and count the interested products in a configurable time window and report that in an output topic.
This sample uses lambda expressions and thus requires Java 8+.
==== Starting Kafka in a docker container
* Skip steps 1-3 if you already have a non-Docker Kafka environment.
1. Go to the docker directory in this repo and invoke the command `docker-compose up -d`.
2. Ensure that in the docker directory and then invoke the script `start-kafka-shell.sh`
3. cd $KAFKA_HOME
4. Start the console producer: +
Assuming that you are running kafka on a docker container on mac osx. Change the zookeeper IP address accordingly otherwise. +
`bin/kafka-console-producer.sh --broker-list 192.168.99.100:9092 --topic products`
5. Start the console consumer: +
Assuming that you are running kafka on a docker container on mac osx. Change the zookeeper IP address accordingly otherwise. +
`bin/kafka-console-consumer.sh --bootstrap-server 192.168.99.100:9092 --key-deserializer org.apache.kafka.common.serialization.IntegerDeserializer --property print.key=true --topic product-counts`
=== Running the app:
Go to the root of the repository and do:
`./mvnw clean package`
`java -jar target/kstream-product-tracker-0.0.1-SNAPSHOT.jar --kstream.product.tracker.productIds=123,124,125 --spring.cloud.stream.kstream.timeWindow.length=60000 --spring.cloud.stream.kstream.timeWindow.advanceBy=30000 --spring.cloud.stream.bindings.input.destination=products`
The above command will track products with ID's 123,124 and 125 every 30 seconds with the counts from the last minute.
In other words, every 30 seconds a new 1 minute window is started.
* By default we use the docker container IP (mac osx specific) in the `application.yml` for Kafka broker and zookeeper.
Change it in `application.yml` (which requires a rebuild) or pass them as runtime arguments as below.
`spring.cloud.stream.kstream.binder.brokers=<Broker IP Address>` +
`spring.cloud.stream.kstream.binder.zkNodes=<Zookeeper IP Address>`
Enter the following in the console producer (one line at a time) and watch the output on the console consumer:
```
{"id":"123"}
{"id":"124"}
{"id":"125"}
{"id":"123"}
{"id":"123"}
{"id":"123"}
```
The default time window is configured for 30 seconds and you can change that using the following property.
`kstream.word.count.windowLength` (value is expressed in milliseconds)
In order to switch to a hopping window, you can use the `kstream.word.count.advanceBy` (value in milliseconds).
This will create an overlapped hopping windows depending on the value you provide.
a2fea3a188