Apache Kafka With Spring Boot and Kotlin

1. Introduction

In this step-by-step guide, I will show you how to use Apache Kafka with Spring Boot and Kotlin.

When you finish this article, you will know precisely:

• how to Produce and Consume String messages,
• how to Publish and Consume JSON messages,
• how to fix an infinite-loop problem with invalid JSON messages

Please keep in mind, that you should have Apache Kafka up and running on your local machine. If you would like to learn how to set it up, then check out my other articles under the Kafka tag.

2. Create New Project

Let’s start with creating a new Spring Boot project. As always, I highly encourage you to use the Spring Initializr page, when creating a new one:

To be on the same page, I’ve selected Spring Boot version 2.7.4 with Jar packaging and Java 17. Additionally, we will need two more dependencies:

• Spring for Apache Kafka– necessary to work with our Kafka node,
• Spring Web– although this one is not necessary, I’ve added it so that we can expose a test REST API endpoint.

[elementor-template id=”9007393″]

3. Publish/Consume String Messages

As the first example, let’s take a look at how to publish and consume simple String messages.

3.1. application.yaml Configuration

After we import our Kafka Spring Boot Kotlin project into our favorite IDE (like IntelliJ, for instance), let’s navigate to the application.yaml file and put the following:

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:8098
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      bootstrap-servers: localhost:8098
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

As we can see, these settings are responsible for setting both Kafka consumer and producer:

• With bootstrap-servers, we point to the host and port, where we can reach out to our Kafka listener. Of course, depending on your node configuration this value may be different.
• When it comes to the auto-offset-reset, it lets us specify consumer behavior when there is no initial offset in Kafka or if the current offset does not exist
• With key-deserializer and value-deserializer, we specify the full qualifier to the deserializer we want to use. As the name suggests, we want to use the StringDeserializer for bot key and value deserialization.
• Finally, the key-serializer and value-serializer let us do almost exactly the same, but for the serialization process.

3.2. Create Constants.kt File

As the next step, let’s create Constants.kt file inside the config package:

const val EXAMPLE_TOPIC_NAME = "someTopicOne"

const val GROUP_ID = "groupId"

As we can see, we put two String constants here with the example topic name and group ID.

Although this is not necessary, it will help us maintain our code a bit cleaner.

3.3. Implement String Messages Producer

Nextly, let’s implement an ExampleStringProducer class inside the producer package:

@Component
class ExampleStringProducer(
  private val kafkaTemplate: KafkaTemplate<String, String>
) {

  fun sendStringMessage(message: String) {
    kafkaTemplate.send(EXAMPLE_TOPIC_NAME, message)
  }

}

As can be seen, in order to send messages we have to inject a KafkaTemplate class. This generic class exposes methods for executing high-level operations and as type parameters (<String, String> in our case), we have to specify the desired key and the value types.

When it comes to the sendStringMessage function, it simply takes a String argument called message and sends it to the topic- called someTopicOne in our case.

3.4 Create Messages Consumer

With that being done, let’s implement a consumer called ExampleConsumer:

@Component
class ExampleConsumer {

  private val logger = LoggerFactory.getLogger(this.javaClass)

  @KafkaListener(topics = [EXAMPLE_TOPIC_NAME], groupId = GROUP_ID)
  fun firstListener(message: String) {
    logger.info("Message received: [$message]")
  }

}

This time, in order to make our firstListener function a target of a Kafka message listener on the specified topics, we have to mark it with the @KafkaListener annotation. Moreover, if we would like to handle multiple topics with this function, we could simply specify their names in the topics array.

It’s worth mentioning that with the groupId, we override the group ID property for the consumer factory, but for this particular listener only.

And to summarize this code snippet- the incoming message will be simply logged into the output.

3.5 Expose Test Controller

Finally, let’s expose a test REST endpoint, which we will use to send messages:

@RestController
class ExampleController(
  private val exampleStringProducer: ExampleStringProducer
) {

  @PostMapping("/test")
  @ResponseStatus(HttpStatus.NO_CONTENT)
  fun sendTestMessage(
    @RequestBody requestBody: RequestBodyDto
  ) {
    exampleStringProducer.sendStringMessage(
      message = requestBody.message
    )
  }

  data class RequestBodyDto(val message: String)
}

As we can see, this handler will be responding to the POST requests to the /test endpoint.

As a request body, we will send the message, which will be then passed to the sendStringMessage function and if everything works fine, we should get a 204 No Content response.

3.6 Test The Endpoint

In order to test our functionality, let’s run a couple of POST requests:

curl --location --request POST 'localhost:8080/test' \
--header 'Content-Type: application/json' \
--data-raw '{
    "message": "SomeMessage"
}'

As a result, we should see our messages printed to the output, like these:

Message received: [Some Example Message]
Message received: [Another one]

// ExampleDto.kt file:
data class ExampleDto(val someMessage: String)

// UserDto.kt file:
data class UserDto(val id: Long, val name: String)

const val EXAMPLE_TOPIC_NAME_TWO = "someTopicTwo"
const val EXAMPLE_TOPIC_NAME_THREE = "someTopicThree"

@Component
class ExampleJsonProducer(
  private val exampleDtoKafkaTemplate: KafkaTemplate<String, ExampleDto>,
  private val userDtoKafkaTemplate: KafkaTemplate<String, UserDto>
) {

  fun sendExampleDtoMessage(dto: ExampleDto) {
    exampleDtoKafkaTemplate.send(EXAMPLE_TOPIC_NAME_TWO, dto)
  }

  fun sendUserDtoMessage(dto: UserDto) {
    userDtoKafkaTemplate.send(EXAMPLE_TOPIC_NAME_THREE, dto)
  }
}

@KafkaListener(topics = [EXAMPLE_TOPIC_NAME_TWO], groupId = GROUP_ID)
fun secondListener(message: ExampleDto) {
  logger.info("Message received: [$message]")
}

@KafkaListener(topics = [EXAMPLE_TOPIC_NAME_THREE], groupId = GROUP_ID)
fun secondListener(message: UserDto) {
  logger.info("Message received: [$message]")
}

@RestController
class ExampleController(
  private val exampleStringProducer: ExampleStringProducer,
  private val exampleJsonProducer: ExampleJsonProducer
)

exampleJsonProducer.sendExampleDtoMessage(
  dto = ExampleDto(requestBody.message)
)
exampleJsonProducer.sendUserDtoMessage(
  dto = UserDto(
    id = Random.nextLong(0, 100),
    name = requestBody.message
  )
)

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:8098
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
    producer:
      bootstrap-servers: localhost:8098
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

curl --location --request POST 'localhost:8080/test' \
--header 'Content-Type: application/json' \
--data-raw '{
    "message": "SomeMessage"
}'

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:8098
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring:
          deserializer:
            value:
              delegate:
                class: org.springframework.kafka.support.serializer.JsonDeserializer
    producer:
      bootstrap-servers: localhost:8098
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

curl --location --request POST 'localhost:8080/test' \
--header 'Content-Type: application/json' \
--data-raw '{
    "message": "SomeMessage"
}'

The class ‘com.codersee.kafkaexample.dto.ExampleDto’ is not in the trusted packages: [java.util, java.lang]. If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).

Similarly, to fix this issue we have to add new property- properties.spring.json.trusted.packages– and either point to the package, where our DTOs live or set a wildcard.

In my case, the final application.yaml file looks, as follows:

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:8098
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring:
          json:
            trusted:
              packages: "com.codersee.kafkaexample.dto"
          deserializer:
            value:
              delegate:
                class: org.springframework.kafka.support.serializer.JsonDeserializer
    producer:
      bootstrap-servers: localhost:8098
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

4.9. Final Testing

Finally, we can rerun our beloved Spring Boot Kafka with Kotlin application and try the request once again:

curl --location --request POST 'localhost:8080/test' \
--header 'Content-Type: application/json' \
--data-raw '{
    "message": "SomeMessage"
}'

As a result, we should see the following:

Message received: [SomeMessage]
Message received: [ExampleDto(someMessage=SomeMessage)]
Message received: [UserDto(id=26, name=SomeMessage)]

As we can see, all 3 Apache Kafka producers and consumers are working, as expected (and if no, then let me know in the comment section).

5. Apache Kafka With Spring Boot and Kotlin Summary

And that would be all for this article about how to work with Apache Kafka With Spring Boot and Kotlin. If you’d like to see the whole project source code, then visit this GitHub repository.

I really hope that you enjoyed this step-by-step guide and will be delighted if you would like to share your feedback with me and the others in the comments section below.

Take care and have a great day! 😀