Trabajo con Kafka en Kotlin: bibliotecas, configuración y ejemplos

Trabajo con Kafka en Kotlin: descripción general de bibliotecas y ejemplos prácticos

Apache Kafka es una plataforma distribuida para transmisión de eventos en flujo, ampliamente utilizada en arquitecturas modernas de microservicios. En el ecosistema de Kotlin, se utilizan dos bibliotecas principales para trabajar con Kafka: Kafka Clients (el cliente oficial de Java de Apache) y Kafka Streams (biblioteca para procesamiento en flujo). Gracias a la compatibilidad total de Kotlin con Java, puede usar estas bibliotecas directamente, así como aplicar envoltorios específicos de Kotlin para un código más conciso.

1. Kafka Clients — trabajo básico con productor y consumidor

Apache Kafka Clients es la biblioteca oficial de Java que proporciona una API de bajo nivel para enviar y recibir mensajes. En Kotlin se conecta a través de Maven o Gradle:

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

Veamos un ejemplo de creación de un productor (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", "Hello from Kotlin!")\n\n    producer.send(record) { metadata, exception ->\n        if (exception != null) {\n            println("Error de envío: ${exception.message}")\n        } else {\n            println("Enviado a la partición ${metadata.partition()} con offset ${metadata.offset()}")\n        }\n    }\n\n    producer.close()\n}

De manera similar se crea un consumidor (consumer) para leer mensajes:

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("Recibido: clave = ${record.key()}, valor = ${record.value()}, offset = ${record.offset()}")\n        }\n    }\n}

2. Kafka Streams — procesamiento de datos en flujo

Kafka Streams es una biblioteca para construir aplicaciones de procesamiento en flujo basadas en Kafka. Permite realizar transformaciones, agregaciones y uniones de flujos de datos directamente en el lado del cliente. Conexión:

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

Ejemplo de una aplicación simple que lee mensajes de un tópico, los convierte a mayúsculas y los escribe en otro tópico:

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