Spring Kafka - how to use ReplyingKafkaTemplate send and reply synchronously

Microservice Spring with Apache Kafka

Kafka is an asynchronous messaging queuing system, but when you have a case in your project that you need to have the result immediately from another service you can do that quite easily.

Kafka from the client side produces a message to the server side, then waiting for the server to return back the message exactly what the client wants. It’s identified in the kafka header request named KafkaHeaders.CORRELATION_ID and a topic where the server should send. If the server side produces the same correlation id to the same topic in the request, then the client side will catch that message as its reply.

Although you can work with kafka from a different framework and language other than Java and Spring. Here’s in this example i just specifically giving the example of sending and replying kafka messages using Spring Boot.

The Project Overview


The client-project will send a string message to the server-project over kafka, then the server-project will reverse the string and return it back to the client.

First let’s start with our pom.xml, for both services we named spring-kafka-client and spring-kafka-server. The dependencies required are as follows.


And also the spring configuration files are the same for both services. Here’s the application.yml that you must create.


1. Client Service


Spring projects the client side, which produces a message and waits for the result.

This is an entry application containing the main method, we annotated them with @EnableKafka.

Main application for spring-kafka-client


Kafka configuration

- We registered bean ReplyingKafkaTemplate into our application, so this kafka template will be available in our application.
- We also register ConcurrentMessageListenerContainer, in this configuration we put REPLY_TOPICS and CONSUMER_GROUPS that are in our Yaml configuration.

Service that produce message


This service will be reusable throughout your client application, it will send synchronous requests and expect results from the producer. Basically we send messages into the SEND_TOPICS topic, with specific payload and headers, then it will be detected by the consumer somewhere on our server project.

2. Server Service


Spring project that will do the process of a given input. It can be some specific business logic, database query, etc.

Main application for spring-kafka-server.


Consumer and producer
- It listens to only a specific topic Which are into topic “${myproject.send-topics}” and by consumer “${myproject.consumer-group}”, those are variables that we set early in properties.
- It then @SendTo the reply topic, Kafka will automatically set the Header and Reply Topic that is sent by the client.
- Then inside the method it’s just a simple stringBuilder reverse string method. You can put your real business logic here.

Test the Application


Because it’s not a rest API web based spring application, we can just do an integration test of our send and reply kafka project, using Junit tests.

Write and run the test.


The test should pass if the server responds with a reversed string, from the message that client is sending. It’s just a simple logic for a test.

You can create a rest API if you want, all you have to do is just autowiring the KafkaService and use the kafkaRequestReply method.

We only do the test in our client project, because we just want to know the replying result. We don't necessarily have to write the test in the server project. But you can write a test for your server(producer) project yourself.

The full project can be clone from this Github repository Spring Kafka Request and wait Reply.

You can find more on the Spring kafka documentation here basically about anything you want to know about Kafka in Spring, here's the spring documentation https://docs.spring.io/spring-kafka/reference/html/#replying-template.


Comments

Popular posts from this blog

ASUS Battery Health Charging Software To Make Your Battery Life Longer

Flutter Button With Left Align Text and Icon

Flutter AppBar, With Icon, Title and Actions Link

ERROR 1348 Column Password Is Not Updatable When Updating MySQL Root Password

How To Create Spring Boot Project Using Netbeans