--- title: Kafka Tutorial with Java Spring Boot published: true date: 2022-09-10 12:25:26 UTC tags: canonical_url: https://rohanshah.hashnode.dev/kafka-tutorial-with-java-spring-boot --- ### Why Kafka? Kafka has become the mainstay of messaging in applications with high-volume requirements According to confluent.io, > Apache Kafka is an open source distributed streaming system used for stream processing, real-time data pipelines, and data integration at scale. This basically means two systems can talk to each other, in a low-to-none overhead and high velocity manner. This article talks about how to make an application (well, actually two) that explains, in the most basic sense, how Kafka works in distributed systems. We have used Spring Boot + Java for this purpose Let us say we distribute our "application" in such a way that one application takes care of serving the REST requests, and takes care of things like _validation_, _data-elimination_ and the second application takes care of saving the data to the database. Here we have made sure that one application will _only_ be catering to REST requests and do some in-memory operations required on the request, the other will always keep listening to the first application and make sure that it saves all the data that the first application will send. We already have **separation of concern** in place here. If you look at our project structures below, one application that caters REST requests does not have a repository, while the other one that saves data does not have a controller. This is how our project structure is going to look like The REST application is called a "producer" since it will be producing the messages to be sent to Kafka, the database application is called a "consumer", since it will be consuming the messages from Kafka ##### Producer application ![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/xugij4nk5tkuhbsneb6j.png) ##### Consumer application ![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/63sb2x2uvogxnjbtnqsm.png) ## Producer Application In our application, we have a REST endpoint called `/user/save` that resides in the Producer application The User model will look something like this: ``` public class User { private String id; private String name; private Integer age; //getters and setters } ``` The Rest controller just has a call to service with the `User` model ``` @RestController @RequestMapping("/user") public class UserController { @Autowired private UserService userService; @PostMapping("/save") public String saveUser(@RequestBody User user) throws JsonProcessingException { return userService.savingUser(user); } } ``` The _Kafkaesque_ magic happens inside the service ``` public String savingUser(User user) throws JsonProcessingException { Properties properties = new Properties(); properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092"); properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); try (KafkaProducer producer = new KafkaProducer<>(properties)) { ObjectMapper mapper = new ObjectMapper(); String userAsString = mapper.writeValueAsString(user); ProducerRecord record = new ProducerRecord<>("MyTopic", userAsString); producer.send(record); } return user.getName(); } ``` In the `Properties` calls, we are just configuring the Kafka properties to the most default and bare minimum Then we initialize a `KafkaProducer`, this is a Default class inside the Kafka dependency. A dependency is required to get this class available at classpath. We have used gradle and this is what the dependency list inside our `build.gradle` looks like: ``` dependencies { implementation 'org.springframework.boot:spring-boot-starter-data-elasticsearch' implementation 'org.springframework.boot:spring-boot-starter-web' implementation 'org.springframework.kafka:spring-kafka' implementation 'org.apache.kafka:kafka-clients' testImplementation 'org.springframework.boot:spring-boot-starter-test' testImplementation 'org.springframework.kafka:spring-kafka-test' } ``` After creation of Producer, we essentially convert the User model to a json String, so it can be sent as a "message" to another system. We then generate a `ProducerRecord` which mentions what the message is (`userAsString`) and where it wants to go (_a Kafka Topic_) Then comes the main call that actually sends this data to the message pipeline, ``` producer.send(record); ``` At this point, we can say that our _message_ has been _queued_ in kafka, and it is ready to be `consumed` **_Exit Producer_** **_Enter Consumer_** ## Consumer Application Now we need a means to consume this message. At this point, - someone has made a request to our rest endpoint to save the user, - the data of the user to be saved has been queued to kafka. Now we will need to actually save this data to a database. This is where the Consumer comes in The heart of this project is `UserConsumer`, a service that consumes the Kafka message that has been waiting in the queue. The Consumer has one and only one method, whose job is to keep listening to Kafka whether or not Kafka has something to say. The run method looks something like this ``` @Override @PostConstruct public void run() { Properties properties = new Properties(); properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092"); properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my-application"); properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); try (KafkaConsumer consumer = new KafkaConsumer<>(properties)) { consumer.subscribe(Collections.singletonList("MyTopic")); System.out.println("INSIDE RUN OF CONSUMER");//should've used a logger while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord record : records) { System.out.println("VALUE:" + record.value()); try { userRepository.save(new ObjectMapper().readValue(record.value(), User.class)); System.out.println(userRepository.findAll().size()); } catch (JsonProcessingException e) { e.printStackTrace(); } } } } } ``` Here, as usual, with the Properties calls, we are bootstrapping our kafka configurations. We then initiate a KafkaConsumer, that subscribes to the topic that the Producer sent the message to, named `MyTopic`. Producer sent a message to topic `MyTopic`, Consumer is listening to topic `MyTopic`. The message sent from our producer will reach the consumer, if everything goes fine. Now we are converting this String message back to our Model class. This Model class has a similar Structure to our Producer's Model class, however it is decorated with some Database specific annotations. ``` @Entity @Table(name = "user_table") public class User { @Id private String id; private String name; private Integer age; // getters and setters } ``` Then, this message is sent to the `repository`'s `save` method, where we have our database logic abstracted. The `Repository` is a basic JPARepository which saves this data into a database. Repository configuration: ``` server.port=8081 spring.datasource.url=jdbc:postgresql://localhost:5432/postgres spring.jpa.hibernate.ddl-auto=update spring.datasource.username= spring.datasource.password= ``` At this point, the User is now saved to the database. ### ...One more thing Though this is a very basic application, the usage of Kafka here gave us a very distinct advantage over a traditional application. Suppose our Consumer application is down, the caller of the API can hit the endpoint, the User data will be queued, and whenever the Consumer gets up again, the message will be read and the User will be saved. Our user got saved despite the database application being down! However, this exercise is left for the user to perform. If you want to try this (these) project(s) on your own, here are the links to the github repository(ies). I will try to keep the ReadMe updated. [https://github.com/imrohanshah/KafkaProducer](https://github.com/imrohanshah/KafkaProducer) [https://github.com/imrohanshah/KafkaConsumer](https://github.com/imrohanshah/KafkaConsumer)

imrohanshah

Kafka Tutorial with Java Spring Boot

Join Our Newsletter. No Spam, Only the good stuff.