Работа с Kafka в Kotlin: библиотеки и примеры кода |…

Работа с Kafka в Kotlin: обзор библиотек и практические примеры

Apache Kafka — это распределённая платформа для потоковой передачи событий, широко используемая в современных микросервисных архитектурах. В экосистеме Kotlin для работы с Kafka применяются две основные библиотеки: Kafka Clients (официальный Java-клиент от Apache) и Kafka Streams (библиотека для потоковой обработки). Благодаря полной совместимости Kotlin с Java, вы можете использовать эти библиотеки напрямую, а также применять Kotlin-специфичные обёртки для более лаконичного кода.

1. Kafka Clients — базовая работа с продюсером и консьюмером

Apache Kafka Clients — это официальная Java-библиотека, которая предоставляет низкоуровневый API для отправки и получения сообщений. В Kotlin она подключается через Maven или Gradle:

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

Рассмотрим пример создания продюсера (producer) на Kotlin:

import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.kafka.clients.producer.ProducerRecord
import 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("Ошибка отправки: ${exception.message}")
        } else {
            println("Отправлено в раздел ${metadata.partition()} с offset ${metadata.offset()}")
        }
    }

    producer.close()
}

Аналогично создаётся консьюмер (consumer) для чтения сообщений:

import org.apache.kafka.clients.consumer.ConsumerRecords
import org.apache.kafka.clients.consumer.KafkaConsumer
import java.time.Duration
import 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("Получено: ключ = ${record.key()}, значение = ${record.value()}, offset = ${record.offset()}")
        }
    }
}

2. Kafka Streams — потоковая обработка данных

Kafka Streams — это библиотека для построения приложений потоковой обработки на основе Kafka. Она позволяет выполнять трансформации, агрегации и соединения потоков данных прямо на стороне клиента. Подключение:

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

Пример простого приложения, которое читает сообщения из топика, преобразует их в верхний регистр и записывает в другой топик:

import org.apache.kafka.streams.KafkaStreams
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.StreamsConfig
import org.apache.kafka.streams.kstream.KStream
import java.util.Properties

fun main() {
    val props = Properties().apply {
        put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-app")
        put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
        put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().javaClass)
        put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().javaClass)
    }

    val builder = StreamsBuilder()
    val sourceStream: KStream<String, String> = builder.stream("input-topic")
    val uppercasedStream = sourceStream.mapValues { value -> value.uppercase() }
    uppercasedStream.to("output-topic")

    val streams = KafkaStreams(builder.build(), props)
    streams.start()

    Runtime.getRuntime().addShutdownHook(Thread {
        streams.close()
    })
}