Travailler avec Kafka en Kotlin : bibliothèques, configuration et exemples

Travailler avec Kafka en Kotlin : aperçu des bibliothèques et exemples pratiques

Apache Kafka est une plateforme distribuée de streaming d'événements, largement utilisée dans les architectures modernes de microservices. Dans l'écosystème Kotlin, deux bibliothèques principales sont utilisées pour travailler avec Kafka : Kafka Clients (le client Java officiel d'Apache) et Kafka Streams (bibliothèque pour le traitement de flux). Grâce à la compatibilité totale de Kotlin avec Java, vous pouvez utiliser ces bibliothèques directement, ainsi que des wrappers spécifiques à Kotlin pour un code plus concis.

1. Kafka Clients — travail de base avec le producteur et le consommateur

Apache Kafka Clients est la bibliothèque Java officielle qui fournit une API de bas niveau pour envoyer et recevoir des messages. En Kotlin, elle s'intègre via Maven ou Gradle :

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

Voici un exemple de création d'un producteur (producer) en 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", "Bonjour depuis Kotlin!")\n\n    producer.send(record) { metadata, exception ->\n        if (exception != null) {\n            println("Erreur d'envoi : ${exception.message}")\n        } else {\n            println("Envoyé à la partition ${metadata.partition()} avec l'offset ${metadata.offset()}")\n        }\n    }\n\n    producer.close()\n}

De même, on crée un consommateur (consumer) pour lire les messages :

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("Reçu : clé = ${record.key()}, valeur = ${record.value()}, offset = ${record.offset()}")\n        }\n    }\n}

2. Kafka Streams — traitement de flux de données

Kafka Streams est une bibliothèque pour construire des applications de traitement de flux basées sur Kafka. Elle permet d'effectuer des transformations, des agrégations et des jointures de flux de données directement côté client. Intégration :

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

Exemple d'une application simple qui lit les messages d'un topic, les convertit en majuscules et les écrit dans un autre topic :

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