40 lines
1.6 KiB
Plaintext
40 lines
1.6 KiB
Plaintext
== 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+.
|
|
|
|
=== Running the app:
|
|
|
|
Go to the root of the repository and do:
|
|
|
|
`docker-compose up -d`
|
|
|
|
`./mvnw clean package`
|
|
|
|
`java -jar target/kafka-streams-product-tracker-0.0.1-SNAPSHOT.jar --app.product.tracker.productIds=123,124,125`
|
|
|
|
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.
|
|
|
|
Issue the following commands:
|
|
|
|
`docker exec -it kafka-prod-tracker /opt/kafka/bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic products`
|
|
|
|
On another terminal:
|
|
|
|
`docker exec -it kafka-prod-tracker /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --key-deserializer org.apache.kafka.common.serialization.IntegerDeserializer --property print.key=true --topic product-counts`
|
|
|
|
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"}
|
|
``` |