Working with Kafka in Kotlin: libraries, setup, and examples

Working with Kafka in Kotlin: library overview and practical examples

Apache Kafka is a distributed event streaming platform widely used in modern microservice architectures. In the Kotlin ecosystem, two main libraries are used for working with Kafka: Kafka Clients (the official Java client from Apache) and Kafka Streams (a library for stream processing). Thanks to Kotlin's full compatibility with Java, you can use these libraries directly, as well as apply Kotlin-specific wrappers for more concise code.

1. Kafka Clients — basic work with producer and consumer

Apache Kafka Clients is the official Java library that provides a low-level API for sending and receiving messages. In Kotlin, it is added via Maven or Gradle:

// build.gradle.ktsdependencies {    implementation("org.apache.kafka:kafka-clients:3.6.0")}

Let's look at an example of creating a producer in Kotlin:

import org.apache.kafka.clients.producer.KafkaProducerimport org.apache.kafka.clients.producer.ProducerRecordimport java.util.Properties

fun main() {    val props = Properties().apply {        put("bootstrap.servers", "localhost:9092")        put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")        put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")    }

    val producer = KafkaProducer<String, String>(props)    val record = ProducerRecord("my-topic", "key1", "Hello from Kotlin!")

    producer.send(record) { metadata, exception ->        if (exception != null) {            println("Send error: ${exception.message}")        } else {            println("Sent to partition ${metadata.partition()} with offset ${metadata.offset()}")        }    }

    producer.close()}

A consumer for reading messages is created similarly:

import org.apache.kafka.clients.consumer.ConsumerRecordsimport org.apache.kafka.clients.consumer.KafkaConsumerimport java.time.Durationimport java.util.Properties

fun main() {    val props = Properties().apply {        put("bootstrap.servers", "localhost:9092")        put("group.id", "my-group")        put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")        put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")        put("auto.offset.reset", "earliest")    }

    val consumer = KafkaConsumer<String, String>(props)    consumer.subscribe(listOf("my-topic"))

    while (true) {        val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))        records.forEach { record ->            println("Received: key = ${record.key()}, value = ${record.value()}, offset = ${record.offset()}")        }    }}

2. Kafka Streams — stream data processing

Kafka Streams is a library for building stream processing applications on top of Kafka. It allows you to perform transformations, aggregations, and joins of data streams directly on the client side. Adding the dependency:

dependencies {    implementation("org.apache.kafka:kafka-streams:3.6.0")}

An example of a simple application that reads messages from a topic, transforms them to uppercase, and writes them to another topic:

import org.apache.kafka.streams.KafkaStreamsimport org.apache.kafka.streams.StreamsBuilderimport org.apache.kafka.streams.StreamsConfigimport org.apache.kafka.streams.kstream.KStreamimport java.util.Properties

fun main() {    val props = Properties().apply {        put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-app")        put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "local