Trabalhando com Kafka em Kotlin: bibliotecas, configuração e exemplos

Trabalhando com Kafka em Kotlin: visão geral das bibliotecas e exemplos práticos

Apache Kafka é uma plataforma distribuída para transmissão de eventos em fluxo, amplamente utilizada em arquiteturas modernas de microsserviços. No ecossistema Kotlin, duas bibliotecas principais são usadas para trabalhar com Kafka: Kafka Clients (cliente Java oficial da Apache) e Kafka Streams (biblioteca para processamento de fluxos). Graças à compatibilidade total do Kotlin com Java, você pode usar essas bibliotecas diretamente, bem como aplicar wrappers específicos do Kotlin para um código mais conciso.

1. Kafka Clients — trabalho básico com produtor e consumidor

Apache Kafka Clients é a biblioteca Java oficial que fornece uma API de baixo nível para enviar e receber mensagens. Em Kotlin, ela é adicionada via Maven ou Gradle:

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

Vejamos um exemplo de criação de um produtor (producer) em 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("Erro ao enviar: ${exception.message}")\n        } else {\n            println("Enviado para a partição ${metadata.partition()} com offset ${metadata.offset()}")\n        }\n    }\n\n    producer.close()\n}

Da mesma forma, um consumidor (consumer) é criado para ler mensagens:

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

2. Kafka Streams — processamento de dados em fluxo

Kafka Streams é uma biblioteca para construir aplicações de processamento de fluxo baseadas em Kafka. Ela permite realizar transformações, agregações e junções de fluxos de dados diretamente no lado do cliente. Adição:

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

Exemplo de uma aplicação simples que lê mensagens de um tópico, converte-as para maiúsculas e as escreve em outro 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