This is the demo repository for the Kafka Summit 2021 talk: Event Streaming with Kafka Streams and Spring Cloud Stream
This is an example of a Spring Cloud Stream processor using Kafka Streams support that shows both KStream and KTable bindings.
Following are the two applications in this repository
-
Spring Cloud Stream based Kafka Streams processor
-
Spring Cloud Stream producer application to generate data for the processor
Kafka Streams processor uses java.util.function.BiFunction to demonstrate two inputs and an output.
The processor consumes user region data as KTable and then user clicks information as KStream.
Then it produces the clicks per region info on the outbound.
The same outbound information is stored in a state store as well to demonstrate interactive query capabilities of Kafka Streams exposed as IteractiveQueryService in Spring Cloud Stream.
The application also has a second processor to listen from the outbound topic to log the information. In addition, the application also exposes a REST endpoint, using which the user clicks data per region can be queried.
From the root of the repo: docker-compose up -d (You can skip it, if you already have Kafka running)
Running the app from an IDE.
Run UserClicksPerRegionApplication from an IDE.
If you want to run it from the CLI:
cd user-clicks-per-region
./mvnw clean package
java -jar target/user-clicks-per-region-0.0.1-SNAPSHOT.jarRun the producer to generate data:
Run UserClicksRegionProducerApplication from an IDE (or from CLI).
This application has two REST endpoints that allow you to publish user-region and user-click information to Kafka topics.
First enter some data for user region.
curl -X POST localhost:8090/user-region/alice/asiaAt this point, Alice lives in Asia.
Now send some click impression data from Alice.
curl -X POST localhost:8090/user-clicks/alice/12Watch the console of the UserClicksPerRegionApplication and see that the clicks per region information is logged from the test processor.
Invoke the REST endpoint to extract this same information through an interactive query.
curl localhost:8080/updates/asia | jq .Enter more POST data as above and verify that you see the correct output.
curl localhost:8080/actuator/metrics | jq .curl localhost:8080/actuator/kafkastreamstopology | jq .
curl localhost:8080/actuator/kafkastreamstopology/clicks-applicationId
curl localhost:8080/actuator/kafkastreamstopology/updates-applicationIdPopular UI tool for visualizing the topology: https://zz85.github.io/kafka-streams-viz/
curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/clicks-in-0Note: All bindings corresponding to this Kafka Streams application id will be stopped.