Kafka #

Apache Kafka is a distributed streaming platform designed to handle real-time data streams at scale. It’s not just a message broker like RabbitMQ — it’s a distributed log that can store data for days and let many consumers read the same data independently. Kafka is used for microservice decoupling, event sourcing, stream processing, audit trails, and data synchronization between systems. In Kotlin, you use Kafka through the Apache Kafka Clients (the official Java library) which works perfectly on the JVM. This article covers Kafka’s main concepts, Producers and Consumers, message serialization, consumer groups, error handling, and patterns used in production.

Kafka’s Main Concepts #

flowchart LR
    P1[Producer 1] --> T[Topic: pesanan]
    P2[Producer 2] --> T
    T --> |Partition 0| C1[Consumer Group A\nEmail Service]
    T --> |Partition 1| C2[Consumer Group A\nEmail Service]
    T --> |Partition 0| C3[Consumer Group B\nInventory Service]
    T --> |Partition 1| C4[Consumer Group B\nInventory Service]

Concept	Explanation
Topic	A message category/channel, like a table in a database
Partition	The division of a topic for parallelism; messages in one partition are ordered
Producer	The party that sends messages to a topic
Consumer	The party that reads messages from a topic
Consumer Group	A set of consumers sharing the load of reading partitions
Offset	A message’s position in a partition, used for progress tracking
Broker	The Kafka server where data is stored

Kafka’s main advantage over traditional message brokers: messages aren’t deleted after being consumed — they’re stored based on the retention period. This allows various consumer groups to read the same data independently.

Setup and Dependencies #

// build.gradle.kts
dependencies {
    // Apache Kafka Clients
    implementation("org.apache.kafka:kafka-clients:3.7.0")

    // kotlinx.serialization for message serialization
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")

    // Coroutines (optional, for async consumers)
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
}

Producer — Sending Messages #

Producer Configuration #

import org.apache.kafka.clients.producer.*
import org.apache.kafka.common.serialization.StringSerializer
import kotlinx.serialization.Serializable
import kotlinx.serialization.json.Json

object KafkaConfig {
    const val BOOTSTRAP_SERVERS = "localhost:9092"

    fun producerProperties(): java.util.Properties {
        return java.util.Properties().apply {
            put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS)
            put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
            put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)

            // Reliability — make sure messages aren't lost
            put(ProducerConfig.ACKS_CONFIG, "all")           // wait for all replicas to acknowledge
            put(ProducerConfig.RETRIES_CONFIG, 3)             // retry on failure
            put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true) // avoid duplicates

            // Performance
            put(ProducerConfig.LINGER_MS_CONFIG, 5)           // wait 5ms for batching
            put(ProducerConfig.BATCH_SIZE_CONFIG, 16384)       // 16KB batch size
            put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy") // compression

            // Identify the application
            put(ProducerConfig.CLIENT_ID_CONFIG, "myapp-producer")
        }
    }
}

Sending a Simple Message #

@Serializable
data class OrderEvent(
    val orderId: String,
    val userId: Long,
    val total: Double,
    val status: String,
    val timestamp: Long = System.currentTimeMillis()
)

class OrderProducer {
    private val json = Json { ignoreUnknownKeys = true }
    private val producer = KafkaProducer<String, String>(KafkaConfig.producerProperties())

    fun sendOrderEvent(event: OrderEvent) {
        val key = event.orderId        // the key determines partition assignment
        val value = json.encodeToString(OrderEvent.serializer(), event)

        val record = ProducerRecord(
            "pesanan-events",  // topic name
            key,
            value
        )

        // Send asynchronously with a callback
        producer.send(record) { metadata, exception ->
            if (exception != null) {
                println("Failed to send message: ${exception.message}")
            } else {
                println(
                    "Message sent — topic: ${metadata.topic()}, " +
                    "partition: ${metadata.partition()}, " +
                    "offset: ${metadata.offset()}"
                )
            }
        }
    }

    // Send and wait for confirmation (synchronous)
    fun sendAndWait(event: OrderEvent): RecordMetadata {
        val record = ProducerRecord(
            "pesanan-events",
            event.orderId,
            json.encodeToString(OrderEvent.serializer(), event)
        )
        return producer.send(record).get()  // .get() blocks until confirmation
    }

    fun close() {
        producer.flush()   // make sure all messages are sent
        producer.close()
    }
}

fun main() {
    val producer = OrderProducer()

    repeat(5) { i ->
        val event = OrderEvent(
            orderId = "ORD-${1000 + i}",
            userId = (i + 1).toLong(),
            total = (i + 1) * 50_000.0,
            status = "CREATED"
        )
        producer.sendOrderEvent(event)
    }

    producer.close()
}

Sending to a Specific Partition #

// By default, Kafka uses a hash of the key to determine the partition
// This ensures all messages with the same key go to the same partition
// → ordering is guaranteed per key

// Send to a specific partition explicitly (rarely needed)
val record = ProducerRecord(
    "pesanan-events",
    0,                // explicit partition number
    "ORD-001",        // key
    jsonPayload       // value
)

Consumer — Reading Messages #

Consumer Configuration #

import org.apache.kafka.clients.consumer.*
import org.apache.kafka.common.serialization.StringDeserializer

fun consumerProperties(groupId: String): java.util.Properties {
    return java.util.Properties().apply {
        put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfig.BOOTSTRAP_SERVERS)
        put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
        put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)

        // Auto-commit offsets every 5 seconds (can be disabled for manual control)
        put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false)  // manual commit is safer

        // Where to start if no offset is stored
        put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")  // or "latest"

        // Polling configuration
        put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100)          // max 100 records per poll
        put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300_000)  // 5 minute polling timeout

        put(ConsumerConfig.CLIENT_ID_CONFIG, "myapp-consumer-$groupId")
    }
}

A Basic Consumer #

class OrderConsumer(groupId: String = "email-service") {
    private val json = Json { ignoreUnknownKeys = true }
    private val consumer = KafkaConsumer<String, String>(consumerProperties(groupId))

    @Volatile private var running = true

    fun start() {
        consumer.subscribe(listOf("pesanan-events"))
        println("Consumer '$groupId' started listening...")

        try {
            while (running) {
                // poll() blocks until there are messages or a timeout
                val records = consumer.poll(java.time.Duration.ofMillis(1000))

                records.forEach { record ->
                    try {
                        processRecord(record)
                    } catch (e: Exception) {
                        println("Error processing record: ${e.message}")
                        // Here you can: log the error, send to a dead letter topic, skip, or throw
                    }
                }

                // Manual commit after all records in the batch are processed
                if (records.count() > 0) {
                    consumer.commitSync()  // or commitAsync() for better performance
                    println("Committed ${records.count()} records")
                }
            }
        } finally {
            consumer.close()
            println("Consumer closed")
        }
    }

    private fun processRecord(record: ConsumerRecord<String, String>) {
        val event = json.decodeFromString(OrderEvent.serializer(), record.value())
        println(
            "Processing order ${event.orderId} | " +
            "partition: ${record.partition()} | " +
            "offset: ${record.offset()} | " +
            "total: Rp${"%,.0f".format(event.total)}"
        )
        // Business logic: send a confirmation email, etc.
        sendConfirmationEmail(event)
    }

    private fun sendConfirmationEmail(event: OrderEvent) {
        println("Confirmation email sent for order ${event.orderId}")
    }

    fun stop() { running = false }
}

A Consumer with Per-Offset Manual Commit #

// Commit offsets only for records that were processed successfully
fun consumerWithSelectiveCommit(consumer: KafkaConsumer<String, String>) {
    consumer.subscribe(listOf("pesanan-events"))

    while (true) {
        val records = consumer.poll(java.time.Duration.ofMillis(1000))

        val offsetsToCommit = mutableMapOf<
            org.apache.kafka.common.TopicPartition,
            OffsetAndMetadata
        >()

        records.forEach { record ->
            try {
                // Process the record
                println("Process: ${record.key()} offset: ${record.offset()}")

                // Record the successful offset
                offsetsToCommit[
                    org.apache.kafka.common.TopicPartition(record.topic(), record.partition())
                ] = OffsetAndMetadata(record.offset() + 1)

            } catch (e: Exception) {
                println("Failed to process record ${record.offset()}: ${e.message}")
                // Stop the loop — commit only up to before the failed record
                return@forEach
            }
        }

        if (offsetsToCommit.isNotEmpty()) {
            consumer.commitSync(offsetsToCommit)
        }
    }
}

Consumer Groups and Parallelism #

Consumer groups allow the read load to be shared among several consumer instances. Each partition is only read by one consumer in the same group:

// Run several consumers in the same group for parallelism
fun main() {
    val threadCount = 3  // must be <= the partition count

    val threads = (1..threadCount).map { i ->
        Thread {
            val consumer = OrderConsumer("email-service")
            // Each consumer will get a different partition
            consumer.start()
        }.apply {
            name = "consumer-thread-$i"
            isDaemon = true
        }
    }

    threads.forEach { it.start() }
    threads.forEach { it.join() }
}

Kafka with Coroutines #

For cleaner integration with async Kotlin code:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// Producer wrapper for coroutines
class CoroutineProducer {
    private val json = Json { ignoreUnknownKeys = true }
    private val producer = KafkaProducer<String, String>(KafkaConfig.producerProperties())

    suspend fun send(topic: String, key: String, value: String) {
        withContext(Dispatchers.IO) {
            val record = ProducerRecord(topic, key, value)
            producer.send(record).get()  // blocking within the IO dispatcher
        }
    }

    suspend fun sendEvent(event: OrderEvent) {
        send(
            "pesanan-events",
            event.orderId,
            json.encodeToString(OrderEvent.serializer(), event)
        )
    }
}

// A consumer as a Flow
fun kafkaFlow(
    topics: List<String>,
    groupId: String,
    scope: CoroutineScope
): Flow<ConsumerRecord<String, String>> = callbackFlow {
    val consumer = KafkaConsumer<String, String>(consumerProperties(groupId))
    consumer.subscribe(topics)

    val job = scope.launch(Dispatchers.IO) {
        try {
            while (isActive) {
                val records = consumer.poll(java.time.Duration.ofMillis(100))
                records.forEach { record ->
                    trySend(record)  // send to the flow
                }
                if (records.count() > 0) consumer.commitAsync()
            }
        } finally {
            consumer.close()
        }
    }

    awaitClose { job.cancel() }
}

// Usage with Flow
fun main() = runBlocking {
    val flow = kafkaFlow(listOf("pesanan-events"), "notification-service", this)
    val json = Json { ignoreUnknownKeys = true }

    flow
        .map { record ->
            json.decodeFromString(OrderEvent.serializer(), record.value())
        }
        .filter { event -> event.status == "CREATED" }
        .collect { event ->
            println("Notification for order ${event.orderId}")
        }
}

Error Handling and Dead Letter Topics #

class ConsumerWithDLT(
    private val groupId: String,
    private val mainTopic: String,
    private val dltTopic: String = "$mainTopic.dlt"
) {
    private val json = Json { ignoreUnknownKeys = true }
    private val consumer = KafkaConsumer<String, String>(consumerProperties(groupId))
    private val producer = KafkaProducer<String, String>(KafkaConfig.producerProperties())

    fun start() {
        consumer.subscribe(listOf(mainTopic))

        while (true) {
            val records = consumer.poll(java.time.Duration.ofMillis(1000))

            records.forEach { record ->
                val success = processWithRetry(record, maxAttempts = 3)

                if (!success) {
                    // Send to the Dead Letter Topic for manual handling
                    sendToDlt(record)
                }
            }

            consumer.commitSync()
        }
    }

    private fun processWithRetry(
        record: ConsumerRecord<String, String>,
        maxAttempts: Int
    ): Boolean {
        repeat(maxAttempts) { attempt ->
            try {
                val event = json.decodeFromString(OrderEvent.serializer(), record.value())
                processEvent(event)
                return true
            } catch (e: Exception) {
                println("Attempt ${attempt + 1}/$maxAttempts failed: ${e.message}")
                if (attempt < maxAttempts - 1) {
                    Thread.sleep(500L * (attempt + 1))  // simple exponential backoff
                }
            }
        }
        return false
    }

    private fun processEvent(event: OrderEvent) {
        // Simulate random failures
        if (Math.random() < 0.1) throw RuntimeException("Failed to process event!")
        println("Event ${event.orderId} processed successfully")
    }

    private fun sendToDlt(record: ConsumerRecord<String, String>) {
        val dltRecord = ProducerRecord(
            dltTopic,
            record.key(),
            record.value()
        ).apply {
            // Add headers for debugging
            headers().add("original-topic", mainTopic.toByteArray())
            headers().add("error-timestamp", System.currentTimeMillis().toString().toByteArray())
            headers().add("consumer-group", groupId.toByteArray())
        }

        producer.send(dltRecord).get()
        println("Record ${record.key()} sent to DLT: $dltTopic")
    }
}

Creating Topics Programmatically #

import org.apache.kafka.clients.admin.*

fun createTopic(name: String, partitions: Int = 3, replication: Short = 1) {
    val properties = java.util.Properties().apply {
        put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfig.BOOTSTRAP_SERVERS)
    }

    AdminClient.create(properties).use { admin ->
        val newTopic = NewTopic(name, partitions, replication).apply {
            configs(mapOf(
                "retention.ms" to "604800000",   // 7 day retention
                "cleanup.policy" to "delete",
                "compression.type" to "snappy"
            ))
        }

        val result = admin.createTopics(listOf(newTopic))
        result.values()[name]?.get()
        println("Topic '$name' created successfully with $partitions partitions")
    }
}

fun listTopics(): Set<String> {
    val properties = java.util.Properties().apply {
        put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfig.BOOTSTRAP_SERVERS)
    }

    return AdminClient.create(properties).use { admin ->
        admin.listTopics().names().get()
    }
}

Exactly-Once Semantics #

For cases needing exactly-once guarantees (no duplicates, no loss):

fun exactlyOnceProducerProperties(): java.util.Properties {
    return KafkaConfig.producerProperties().apply {
        put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true)
        put(ProducerConfig.ACKS_CONFIG, "all")
        put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "myapp-producer-1")  // unique ID per instance
    }
}

fun sendWithTransaction(producer: KafkaProducer<String, String>, events: List<OrderEvent>) {
    val json = Json { ignoreUnknownKeys = true }
    producer.initTransactions()

    try {
        producer.beginTransaction()

        events.forEach { event ->
            producer.send(
                ProducerRecord(
                    "pesanan-events",
                    event.orderId,
                    json.encodeToString(OrderEvent.serializer(), event)
                )
            )
        }

        producer.commitTransaction()
        println("${events.size} events sent successfully in one transaction")

    } catch (e: Exception) {
        producer.abortTransaction()
        println("Transaction aborted: ${e.message}")
        throw e
    }
}

Summary #

Kafka for decoupling and streaming — use Kafka when services need to communicate without tight coupling, or when data needs to be consumed by many services independently.
The key determines the partition — all messages with the same key go to the same partition, guaranteeing ordering per entity. Use entity IDs (order IDs, user IDs) as keys.
ACKS=all and idempotence — for production, always set acks=all and enable.idempotence=true to avoid message loss and duplication.
Manual commits are safer than auto-commit — with enable.auto.commit=false and commitSync() after processing, you’re sure offsets are only committed after messages are successfully processed.
Consumer groups for parallelism — the number of consumers in a group must not exceed the partition count. Add partitions when you need to increase parallelism.
Dead Letter Topics for error handling — messages that fail after retries are sent to a DLT for manual handling. Add informative headers (error message, timestamp, original topic).
Partitions = the unit of parallelism — create topics with a partition count matching your throughput predictions. It’s easier to add partitions than to reduce them (reducing can change routing).
Kafka isn’t a database replacement — Kafka is a distributed log for streaming, not a database. Use Kafka alongside a database (PostgreSQL, MongoDB) in event-driven architectures.

← Previous: Elasticsearch Next: RabbitMQ →