Este documento detalla el proceso de instalación y configuración de Apache Kafka en un entorno Windows, junto con ejemplos prácticos de cómo utilizarlo para enviar y recibir mensajes mediante Java.

Instalación de Kafka en Windows

Para comenzar, descargue la última versión binaria de Kafka desde el sitio oficial: http://kafka.apache.org/downloads.html.

Extracción de Archivos

Una vez descargado, descomprima el archivo en una ubicación de fácil acceso en su disco duro. Se recomienda evittar rutas de directorio excesivamente profundas (por ejemplo, D:\kafka\_2.12-2.5.0). Rutas muy largas pueden causar problemas al iniciar los servicios de ZooKeeper o Kafka, manifestándose en errores como "input line too long; the syntax of the command is incorrect".

Configuración de Parámetros

Modifique los siguientes archivos de configuración dentro del directorio config:

• config/zookeeper.properties: Ajuste el parámetro dataDir para especificar la ruta donde ZooKeeper almacenará sus datos. Cree este directorio si no existe. Ejemplo: dataDir=D:\\kafka\_2.12-2.5.0\\dataDir
• config/server.properties: Configure el parámetro log.dirs para indicar la ubicación de los logs de Kafka. Ejemplo: log.dirs=D:\\kafka\_2.12-2.5.0\\logs

Inicio de Servicios

Los scripts para iniciar ZooKeeper y Kafka se encuentran en el subdirectorio bin/windows de su instalación de Kafka. Tenga en cuenta que estos son específicos para Windows.

Iniciar ZooKeeper:

Ejecute el siguiente comando en la consola, asegurándose de estar en el directorio bin/windows:

.\zookeeper-server-start.bat ..\config\zookeeper.properties

Iniciar Kafka:

Una vez que ZooKeeper esté en funcionamiento, inicie el servidor Kafka con el siguiente comando:

.\kafka-server-start.bat ..\config\server.properties

Uso de Kafka con Java

Para interactuar con Kafka desde aplicaciones Java, necesitará agregar la dependencia correspondiente a su proyecto. Si utiliza Maven, incluya lo siguiente en su pom.xml:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.5.0</version> <!-- Asegúrese de que la versión coincida con su instalación de Kafka -->
</dependency>

Productor (Producer)

El siguiente código demuestra cómo crear un productor que envía mensajes a un tópico específico de Kafka.

package com.example.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ExecutionException;

public class MessageProducer {

    private static final String KAFKA_TOPIC = "mi-topico-kafka";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; // Ajuste según su configuración

    public static void main(String[] args) {
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
        // Utilice la forma generica sin tipos para evitar advertencias de versión,
        // o especifique los tipos si la versión de la librería lo requiere explícitamente.
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)) {
            System.out.println("Iniciando envío de mensajes...");
            while (true) {
                String messageContent = "Mensaje #" + new Random().nextInt(1000);
                ProducerRecord<String, String> record = new ProducerRecord<>(KAFKA_TOPIC, messageContent);

                // Enviar el mensaje de forma asíncrona y esperar la confirmación
                producer.send(record).get(); // .get() hace la llamada síncrona para simplificar el ejemplo

                System.out.println("Mensaje enviado: " + messageContent);
                Thread.sleep(1500); // Pausa entre envíos
            }
        } catch (InterruptedException | ExecutionException e) {
            System.err.println("Error al enviar el mensaje: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }
}

Consumidor (Consumer)

El siguiente código implementa un consumidor que se suscribe a un tópico y procesa los mensajes recibidos.

package com.example.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class MessageConsumer {

    private static final String KAFKA_TOPIC = "mi-topico-kafka";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; // Ajuste según su configuración
    private static final String CONSUMER_GROUP_ID = "grupo-consumidores-1";

    public static void main(String[] args) {
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // Configuración para re-leer mensajes si es necesario (ej: "earliest", "latest")
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        // Utilice la forma generica sin tipos para evitar advertencias de versión,
        // o especifique los tipos si la versión de la librería lo requiere explícitamente.
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singletonList(KAFKA_TOPIC));
            System.out.println("Esperando mensajes en el tópico: " + KAFKA_TOPIC);

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                records.forEach(record ->
                    System.out.printf("Tópico: %s, Partición: %d, Offset: %d, Mensaje: %s%n",
                            record.topic(), record.partition(), record.offset(), record.value())
                );
            }
        } catch (Exception e) {
            System.err.println("Error en el consumidor: " + e.getMessage());
        }
    }
}

Verificación

Tras iniciar el productor y el consumidor, debería observar que los mensajes enviados por el productor son recibidos y mostrados en la consola por el consumidor.