In order for our application to be able to communicate with Kafka, we’ll need to define an outbound stream to write messages to a Kafka topic, and an inbound stream to read messages from a Kafka topic.

Spring Cloud provides a convenient way to do this by simply creating an interface that defines a separate method for each stream.

The inboundGreetings() method defines the inbound stream to read from Kafka and outboundGreetings() method defines the outbound stream to write to Kafka.

During runtime Spring will create a java proxy based implementation of the GreetingsStreams interface that can be injected as a Spring Bean anywhere in the code to access our two streams.

Configure Spring Cloud Stream

Our next step is to configure Spring Cloud Stream to bind to our streams in the GreetingsStreams interface. This can be done by creating a @Configuration class com.kaviddiss.streamkafka.config.StreamsConfig with below code:

The above configuration properties configure the address of the Kafka server to connect to, and the Kafka topic we use for both the inbound and outbound streams in our code. They both must use the same Kafka topic!

Notice how the class doesn’t have any getters and setters thanks to the Lombok annotations. The @ToString will generate a toString() method using the class’ fields and the @Builder annotation will allow us creating Greetings objects using fluent builder (see below).

Create service layer to write to Kafka

Let’s create the com.kaviddiss.streamkafka.service.GreetingsService class with below code that will write a Greetingsobject to the greetings Kafka topic:

The @RestController annotation tells Spring that this is a Controller bean (the C from MVC). The greetings() method defines an HTTP GET /greetings endpoint that takes a message request param and passes it to the sendGreeting() method in GreetingsService.

Listening on the greetings Kafka topic

Let’s create a com.kaviddiss.streamkafka.service.GreetingsListener class that will listen to messages on the greetingsKafka topic and log them on the console:

The @Component annotation similarly to @Service and @RestController defines a Spring Bean.

GreetingsListener has a single method, handleGreetings() that will be invoked by Spring Cloud Stream with every new Greetings message object on the greetings Kafka topic. This is thanks to the @StreamListener annotation configured for the handleGreetings() method.

Running the application

The last piece of the puzzle is the com.kaviddiss.streamkafka.StreamKafkaApplication class that was auto-generated by the Spring Initializer: