Arbeit mit Kafka in Kotlin: Bibliotheken, Einrichtung und Beispiele

Arbeit mit Kafka in Kotlin: Überblick über Bibliotheken und praktische Beispiele

Apache Kafka ist eine verteilte Plattform für die ereignisgesteuerte Datenstromübertragung, die in modernen Microservice-Architekturen weit verbreitet ist. Im Kotlin-Ökosystem werden für die Arbeit mit Kafka zwei Hauptbibliotheken verwendet: Kafka Clients (der offizielle Java-Client von Apache) und Kafka Streams (eine Bibliothek für die Stream-Verarbeitung). Dank der vollständigen Kompatibilität von Kotlin mit Java können Sie diese Bibliotheken direkt nutzen sowie Kotlin-spezifische Wrapper für prägnanteren Code anwenden.

1. Kafka Clients – grundlegende Arbeit mit Producer und Consumer

Apache Kafka Clients ist die offizielle Java-Bibliothek, die eine Low-Level-API zum Senden und Empfangen von Nachrichten bereitstellt. In Kotlin wird sie über Maven oder Gradle eingebunden:

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

Betrachten wir ein Beispiel zur Erstellung eines Producers in Kotlin:

import org.apache.kafka.clients.producer.KafkaProducer\nimport org.apache.kafka.clients.producer.ProducerRecord\nimport java.util.Properties\n\nfun main() {\n    val props = Properties().apply {\n        put("bootstrap.servers", "localhost:9092")\n        put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")\n        put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")\n    }\n\n    val producer = KafkaProducer<String, String>(props)\n    val record = ProducerRecord("my-topic", "key1", "Hello from Kotlin!")\n\n    producer.send(record) { metadata, exception ->\n        if (exception != null) {\n            println("Fehler beim Senden: ${exception.message}")\n        } else {\n            println("Gesendet an Partition ${metadata.partition()} mit Offset ${metadata.offset()}")\n        }\n    }\n\n    producer.close()\n}

Analog dazu wird ein Consumer zum Lesen von Nachrichten erstellt:

import org.apache.kafka.clients.consumer.ConsumerRecords\nimport org.apache.kafka.clients.consumer.KafkaConsumer\nimport java.time.Duration\nimport java.util.Properties\n\nfun main() {\n    val props = Properties().apply {\n        put("bootstrap.servers", "localhost:9092")\n        put("group.id", "my-group")\n        put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")\n        put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")\n        put("auto.offset.reset", "earliest")\n    }\n\n    val consumer = KafkaConsumer<String, String>(props)\n    consumer.subscribe(listOf("my-topic"))\n\n    while (true) {\n        val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))\n        records.forEach { record ->\n            println("Empfangen: Schlüssel = ${record.key()}, Wert = ${record.value()}, Offset = ${record.offset()}")\n        }\n    }\n}

2. Kafka Streams – Stream-Verarbeitung von Daten

Kafka Streams ist eine Bibliothek zur Erstellung von Stream-Verarbeitungsanwendungen auf Basis von Kafka. Sie ermöglicht Transformationen, Aggregationen und Verknüpfungen von Datenströmen direkt auf der Client-Seite. Einbindung:

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

Beispiel einer einfachen Anwendung, die Nachrichten aus einem Topic liest, in Großbuchstaben umwandelt und in ein anderes Topic schreibt:

import org.apache.kafka.streams.KafkaStreams\nimport org.apache.kafka.streams.StreamsBuilder\nimport org.apache.kafka.streams.StreamsConfig\nimport org.apache.kafka.streams.kstream.KStream\nimport java.util.Properties\n\nfun main() {\n    val props = Properties().apply {\n        put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-app")\n        put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "local