{{announcement.body}}
{{announcement.title}}

Producer Consumer With Kafka and Kotlin

DZone 's Guide to

Producer Consumer With Kafka and Kotlin

In this article, we walk through how to develop a simple Spring Boot application using Kafka and Kotlin.

· Big Data Zone ·
Free Resource

Introduction

In this article, we will develop a simple Spring Boot application using Kafka and Kotlin. 

Let's get started. Visit https://start.spring.io and add the following dependencies:

Groovy


This demo makes use of Gradle as the build option. You can choose Maven as well.

Generate and download the project. Then, import this project into IntelliJ IDEA.

Download Apache Kafka

Download the latest version of Apache Kafka from its site and unzip it to a folder. I am using Windows 10 OS. So, you may run into some problem when starting Kafka. It is related to "too many lines encountered". This is because Kafka is appending a big folder structure as the name for its path. If this problem persists, you will have to rename the folder structure to a shorter one and start the application from 'Power Shell'

 The following commands are used to start the Kafka: 

Shell


You can see these two commands in '/bin/windows' folder. 

In order to run Kafka, you need to start the Zookeeper service first. ZooKeeper is an Apache product that offers distributed configuration service. 

Spring Boot Starter

First step is to create a class called KafkaDemoApplication.kt using your IDE. When you created a project from the Spring Starter web site, this class will be created automatically.

Add the following lines of code:

Kotlin


Producer

We can send the message to the topic in two ways, which are listed below.

Next, we need to develop a controller class, which is used to send and receive the message. Let call this class KafkaController.kt. Then, add the following method:

Kotlin


We are using KafkaTemplate to send the message to a topic called test_topic. This will return a ListenableFuture object from which we can get the result of this action. This approach is the easiest one if  you just want to send a message to a topic.

Another Method

The next method of sending a message to Kafka topic is to use the KafkaProducer object.  We will develop that piece of code.

Kotlin

 

It deserves some explanation.

We need to initialize the KafkaProduce object with a Map that contains a key and value for serialization. In this example, we are dealing with string  message so that we need to use only StringSerializer.  

Basically, a Serializer is an interface in Kafka which will convert a string to bytes. Apache Kafka has other serializers, such as ByteArraySerializer, ByteSerializer, FloatSerializer, etc. 

We specify the key and value of the map with the StringSerializer

Kotlin


The next value is the bootstrap server details that is required to communicate with the Kafka cluster.

Kotlin


All these three attributes are necessary if we use KafkaProducer

Then, we need create a ProducerRecord with the name of the topic and the message itself. This is what is achieved in this line.

Kotlin


Now we can send our message to the topic using the following code:

Kotlin


This operation will return a future with the name of the topic that is used to send the message.

Consumer

We have  seen how to send a message to a topic. But we need to listen for the incoming message. In order to achieve this, we need a develop a listener so that we can consume the message.

Let's create a class called MessageConsumer.kt and annotate with the @Service annotation.

Kotlin

 
This method is used to listen for the message with the help of the @KafkaListener annotation and prints the message on the console once it is available in the topic. But make sure you use the same topic name that is used to send the message to.

You can check the whole source code in my github link repository.

Topics:
kafka, kafka apache, kotlin, kotlin and spring, spring, spring boot

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}