== Spring Cloud Stream and Kafka Streams

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.

=== Running the app:

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.jar

Run 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/asia

At this point, Alice lives in Asia.

Now send some click impression data from Alice.

curl -X POST localhost:8090/user-clicks/alice/12

Watch 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.

Accessing binder health endpoint

curl localhost:8080/actuator/health | jq .

Accessing Kafka Streams metrics

curl localhost:8080/actuator/metrics | jq .

Something more specific

curl localhost:8080/actuator/metrics/kafka.stream.thread.commit.total | jq .

Visualize Kafka Streams topology

curl localhost:8080/actuator/kafkastreamstopology | jq .
curl localhost:8080/actuator/kafkastreamstopology/clicks-applicationId
curl localhost:8080/actuator/kafkastreamstopology/updates-applicationId

Popular UI tool for visualizing the topology: https://zz85.github.io/kafka-streams-viz/

Accessing all the bindings

curl localhost:8080/actuator/bindings | jq .

Stopping binding

curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/clicks-in-0

Note: All bindings corresponding to this Kafka Streams application id will be stopped.

Starting binding

curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/clicks-in-0