Setting up Spring Kafka With Kotlin
Tags: spring, kafka, java, kotlin
Published: 2022-02-08 01:34:33

## Creating Spring project

- **Consumer**: https://github.com/magnuspedro/KafkaConsumer
- **Producer**: https://github.com/magnuspedro/KafkaProducer

### Description

This tutorial is going to show how to create a spring Kotlin application from scratch to create a Kafka consumer and producer and set up Kafka with docker. We also going to user Sleuth to trace our message over the services, so we can keep a track of each transaction.

## Installation

To create a new spring project I'm going to use the [start.spring.io](https://start.spring.io/) with Kotlin.

The first service is going to be the producer as the following image:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/stehr8v5weiuiwqn5m3e.png)

The consumer witch is going to be our second sever is going to have the same configurations as the producer, but with a consumer on its name.

We are going to use [sleuth](https://spring.io/projects/spring-cloud-sleuth) because it traces all our requests over the services, and prints them on our logs.

now we are installing the `logback` lib to the project for logging. All we have to do is add the libs to the `pom.xml` so maven will know we want it as a dependency, and run `nvm clean install` to install it.
 
```xml
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-core</artifactId>
    <version>1.2.6</version>
</dependency>
```

### Setting Up Kafka with Docker

In the first step, we are going to set up Kafka with docker, using a `docker-compose.yml` file.

```yaml
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 22181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
```
The first service to be installed with our `docker-compose` is the zookeeper. The zookeeper allows you to create a Kafka cluster, and it's going to take care of the connection and distribution. The second service is Kafka itself.

##### Configuration
- **PLAINTEXT**: We are setting the protocols to [plaintext](https://collectd.org/wiki/index.php/Plain_text_protocol), which means that it's non-encrypted messages. It's not recommended for production, but it's alright for our tests.
- **Kafka Exposed Port**: 29092
- **Zookeeper Exposed Port**: 22181
- **Kafka Replication Factor**: The replication factor is the number of clusters that the partitions are going to be replicated, in our case only on our main cluster 1. 

For a better understanding of the variables and Kafka itself, I would recommend [this article](https://docs.confluent.io/platform/current/kafka/multi-node.html)

## Configuring The Producer

To be able to use produce Kafka messages we first need to implement a few configurations so we can send them to our Kafka server.

### Properties Config

First, we are going to change the `application.properties` file to an `application.yml`, just because I like it better if you want you can keep the `application.properties`.

For the properties configurations, we are going to create a variable to the Kafka address and topic.

```yml
Kafka:
  bootstrapAddress: localhost:29092
  topics:
    product: product
```
We are using the `localhost:29092` as it was set on our container.

### Kafka Config

For the Kafka config file, we will use the address to create a Map that has our server address and then create a new topic with the name we defined on our `application.yml` file.
As we are creating a `@Configuration` it's going to be injected on running time to our code, so after spring finishing booting the topic is going to be created.

```Kotlin
import org.apache.kafka.clients.admin.AdminClientConfig
import org.apache.kafka.clients.admin.NewTopic
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.core.KafkaAdmin

@Configuration
class KafkaConfig(
    @Value("\${kafka.bootstrapAddress}")
    private val servers: String,
    @Value("\${kafka.topics.product}")
    private val topic: String
) {

    @Bean
    fun kafkaAdmin(): KafkaAdmin {
        val configs: MutableMap<String, Any?> = HashMap()
        configs[AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG] = servers
        return KafkaAdmin(configs)
    }

    @Bean
    fun porduto(): NewTopic {
        return NewTopic(topic, 1, 1.toShort())
    }
}
```
#### Producer Config

Now we have the topic configuration, but we need to produce messages to post on the topic, for that we are going to create a producer.

```Kotlin
import com.magnuspedro.kafka.producer.kafkaProducer.config.serializer.ProductSerializer
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.serialization.StringSerializer
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.core.ProducerFactory


@Configuration
class KafkaProducerConfig(
    @Value("\${kafka.bootstrapAddress}")
    private val servers: String
) {
    @Bean
    fun producerFactory(): ProducerFactory<String, Any> {
        val configProps: MutableMap<String, Any> = HashMap()
        configProps[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = servers
        configProps[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
        configProps[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = ProductSerializer::class.java
        return DefaultKafkaProducerFactory(configProps)
    }

    @Bean
    fun kafkaTemplate(): KafkaTemplate<String, Any> {
        return KafkaTemplate(producerFactory())
    }
}
```
The producer has even more configuration, it takes the server address, and two serializes. The key serializer is represented by the `KEY_SERIALIZER_CLASS_CONFIG` that is going to be a string and the `VALUE_SERIALIZER_CLASS_CONFIG` that is a custom serializer created to convert our product entity into JSON.

We are creating a `KafkaTemplate` implementation with our configured producer factory.

#### Custom serializer

For our custom serializer, we are going to use Jackosn to write a byte array from our entity. It has to be a byte array because we are implementing the Serializer Class from spring Kafka.

```Kotlin
import com.fasterxml.jackson.databind.ObjectMapper
import com.magnuspedro.kafka.producer.kafkaProducer.entities.Product
import org.apache.kafka.common.errors.SerializationException
import org.apache.kafka.common.serialization.Serializer
import org.slf4j.LoggerFactory


class ProductSerializer : Serializer<Product> {
    private val objectMapper = ObjectMapper()
    private val log = LoggerFactory.getLogger(javaClass)

    override fun serialize(topic: String?, data: Product?): ByteArray? {
        log.info("Serializing...")
        return objectMapper.writeValueAsBytes(
            data ?: throw SerializationException("Error when serializing Product to ByteArray[]")
        )
    }

    override fun close() {}
}
```
For the serialization, we are only accepting data that is not null, or else we are going to throw a `SerializationException`.

Our Product file is going to be a simple Kotlin data class

```Kotlin
import com.fasterxml.jackson.annotation.JsonProperty

data class Product(
    @JsonProperty("name")
    val name: String,
    @JsonProperty("sku")
    val sku: String?
)
```

We are adding the `@JsonProperty` annotation so Jackson knows the name of each field when it's going to convert it.

#### Test Controller

At last, we need something to send messages to the topic, I have chosen to create a controller, so we can change our objects and send as many as we want. 

```Kotlin
import com.magnuspedro.kafka.producer.kafkaProducer.entities.Product
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Value
import org.springframework.http.HttpStatus
import org.springframework.http.ResponseEntity
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.support.KafkaHeaders
import org.springframework.messaging.Message
import org.springframework.messaging.support.MessageBuilder
import org.springframework.validation.annotation.Validated
import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RequestMapping
import org.springframework.web.bind.annotation.RestController


@RestController
@RequestMapping("/product")
class LogController(
    @Value("\${kafka.topics.product}") val topic: String,
    @Autowired
    private val kafkaTemplate: KafkaTemplate<String, Any>
) {
    private val log = LoggerFactory.getLogger(javaClass)

    @PostMapping
    fun post(@Validated @RequestBody product: Product): ResponseEntity<Any> {
        return try {
            log.info("Receiving product request")
            log.info("Sending message to Kafka {}", product)
            val message: Message<Product> = MessageBuilder
                .withPayload(product)
                .setHeader(KafkaHeaders.TOPIC, topic)
                .setHeader("X-Custom-Header", "Custom header here")
                .build()
            kafkaTemplate.send(message)
            log.info("Message sent with success")
            ResponseEntity.ok().build()
        } catch (e: Exception) {
            log.error("Exception: {}",e)
            ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Error to send message")
        }
    }
}
```
On the controller, we are going to create a `Message` type, so we can add Kafka headers and custom headers when sending the message to the topic. We are going to add only one custom header and see if it's going to come back to us on our producer.

That finishes all the configurations we need to implement a simple Kafka producer, to send JSON messages.

## Kafka Consumer

For the Kafka consumer, we are going to need the same basic configurations as our producer, we need to change our `application.properties` to `application.yml` and create the same `Product` entity, so we can change it back from JSON.

> If you are going to run both projects on the same machine you need to change on project port, to do so you need to add this config to the `application.yml` file.

```yml
logging:
  level:
    root: info
```

#### Kafka Consumer Config

Our consumer config file is going to be similar to our producer.

```Kotlin
import com.magnuspedro.kafka.consumer.KafkaConsumer.config.deserializer.ProductDeserializer
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.annotation.EnableKafka
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.listener.ContainerProperties


@EnableKafka
@Configuration
class KafkaConsumerConfig(
    @Value("\${kafka.bootstrapAddress}")
    private val servers: String
) {

    @Bean
    fun consumerFactory(): ConsumerFactory<String?, Any?> {
        val props: MutableMap<String, Any> = HashMap()
        props[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = servers
        props[ConsumerConfig.GROUP_ID_CONFIG] = "ppr"
        props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
        props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = ProductDeserializer::class.java
        props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
        return DefaultKafkaConsumerFactory(props)
    }

    @Bean
    fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, Any>? {
        val factory = ConcurrentKafkaListenerContainerFactory<String, Any>()
        factory.consumerFactory = consumerFactory()
        factory.containerProperties.ackMode = ContainerProperties.AckMode.MANUAL_IMMEDIATE
        factory.containerProperties.isSyncCommits = true;
        return factory
    }
}
```
Most of the configuration as said before is like our producer, and like it, we have a custom deserializer for the value.

##### Configuration

> A committed message is a message that has already been read and confirmed to be consumed.

- **GROUP_ID_CONFIG**: Is the identification of the group that the consumer belongs
- **AUTO_OFFSET_RESET_CONFIG**: Set the configuration when Kafka consumer starts and has no committed messages, `earliest` meaning that it's going the get all the menages from the beginning and `latest` is going to get only the last message
- **AckMode.MANUAL_IMMEDIATE**: Is setting the commit of the messages to be done manually

#### Custom Deserializer

For our custom deserializer, we are going to implement the Deserializer Class from Spring Kafka and override the `deserialize` method.

```Kotlin
import com.fasterxml.jackson.databind.ObjectMapper
import com.magnuspedro.kafka.consumer.KafkaConsumer.entities.Product
import org.apache.kafka.common.errors.SerializationException
import org.apache.kafka.common.serialization.Deserializer
import org.slf4j.LoggerFactory
import kotlin.text.Charsets.UTF_8


class ProductDeserializer : Deserializer<Product> {
    private val objectMapper = ObjectMapper()
    private val log = LoggerFactory.getLogger(javaClass)

    override fun deserialize(topic: String?, data: ByteArray?): Product? {
        log.info("Deserializing...")
        return objectMapper.readValue(
            String(
                data ?: throw SerializationException("Error when deserializing byte[] to Product"), UTF_8
            ), Product::class.java
        )
    }

    override fun close() {}

}
```

For the deserialization, we are only going to convert the byte array that we received to a string with the `UTF_8` char-set and use Jackson to convert back to our entity. As the serializer, we are going to throw an error if the data is null.

### Kafka Consumer

For consuming messages we first need to create a Consumer, and for that Spring Kafka offers an annotation that creates a listener for the topic.

```Kotlin
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.slf4j.LoggerFactory
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.kafka.support.Acknowledgment
import org.springframework.stereotype.Component

@Component
class Consumer {
    private val logger = LoggerFactory.getLogger(javaClass)

    @KafkaListener(topics = ["\${kafka.topics.product}"], groupId = "ppr")
    fun listenGroupFoo(consumerRecord: ConsumerRecord<Any, Any>, ack: Acknowledgment) {
        logger.info("Message received {}", consumerRecord)
        ack.acknowledge()
    }
}
```
After annotating our method with `KafkaListener` we are going to log the `consumerRecord` and acknowledge (commit the message), so we only consume once each message.

## Results

Now for the results, I'm going to run both applications and send a post request with the JSON object using insomnia to the producer service that is going to send the message to Kafka and it's going to be consumed by the consumer.

![insomnia request](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/gjqbx0bvihwjoi8ae090.png)

Now we have the producer logs showing us that the message was sent to the Topic


![Producer Logs](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/n2cr36672usrthb5ndm4.png)

Pay attention to the Trace ID `b79dc0a696fa6550` that is going to be the same as even though we are receiving the message from another service

![Consumer Logs](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/1wou8bc3tf2jsjxxgyax.png)

The message is too big to fit on my monitor so I'm going to pass it here as plain text

```
2022-02-07 22:27:09.178  INFO [,b79dc0a696fa6550,d95cf5541abb5fb9] 83720 --- [ntainer#0-0-C-1] c.m.k.c.K.infra.consumer.Consumer        : Message received ConsumerRecord(topic = product, partition = 0, leaderEpoch = 0, offset = 104, CreateTime = 1644283629123, serialized key size = -1, serialized value size = 42, headers = RecordHeaders(headers = [RecordHeader(key = X-Custom-Header, value = [67, 117, 115, 116, 111, 109, 32, 104, 101, 97, 100, 101, 114, 32, 104, 101, 114, 101]), RecordHeader(key = spring_json_header_types, value = [123, 34, 88, 45, 67, 117, 115, 116, 111, 109, 45, 72, 101, 97, 100, 101, 114, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 125])], isReadOnly = false), key = null, value = Product(name=Test for Dev.To, sku=1232343))
```

There we have it, our message was propagated to our Kafka consumer and converted from JSON to our Product entity.