Instalación de Apache Kafka en Linux

Apache Kafka es una plataforma de streaming de eventos distribuida diseñada para entrega de mensajes de alto rendimiento y baja latencia. Utiliza un modelo publicar-suscribirse con almacenamiento persistente, lo que la hace ideal para construir canales de datos en tiempo real y aplicaciones de streaming. Esta guía cubre instalación, configuración y operaciones básicas para ejecutar Kafka en Linux.

Tabla de Contenidos

Requisitos Previos

Antes de instalar Kafka, asegúrate de tener:

  • Sistema Linux (Ubuntu 20.04+, CentOS 8+, o Debian 11+)
  • Acceso root o sudo
  • Al menos 4GB de RAM disponible
  • Al menos 50GB de espacio en disco para almacenamiento de mensajes
  • Java 8 o superior instalado
  • Conocimiento básico de redes

Instalación de Java y Dependencias

Kafka requiere Java para ejecutarse. Instala el tiempo de ejecución de OpenJDK:

sudo apt-get update
sudo apt-get install -y openjdk-11-jdk

Verifica la instalación de Java:

java -version

La salida debe mostrar la versión 11 de Java o superior.

Para CentOS/RHEL:

sudo yum install -y java-11-openjdk java-11-openjdk-devel

Instalación de Apache Kafka

Descarga la última versión de Kafka del repositorio de Apache. Verifica las versiones actuales en https://kafka.apache.org/downloads:

cd /opt
sudo wget https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgz
sudo tar -xzf kafka_2.13-3.6.1.tgz
sudo mv kafka_2.13-3.6.1 kafka
sudo chown -R kafka:kafka /opt/kafka

Crea un usuario dedicado para Kafka:

sudo useradd -r -s /bin/bash kafka

Establece variables de entorno en el perfil del usuario de Kafka:

sudo tee /home/kafka/.bashrc <<EOF
export KAFKA_HOME=/opt/kafka
export PATH=$KAFKA_HOME/bin:$PATH
EOF

Para acceso en todo el sistema, crea un enlace simbólico:

sudo ln -s /opt/kafka/bin/kafka-* /usr/local/bin/

ZooKeeper vs Modo KRaft

Kafka tradicionalmente requiere Apache ZooKeeper para la coordinación del cluster. Sin embargo, Kafka 3.3+ soporta el modo KRaft (Kafka Raft), eliminando la dependencia de ZooKeeper.

Uso de ZooKeeper (Método Tradicional)

Instala ZooKeeper:

sudo apt-get install -y zookeeper

Inicia ZooKeeper:

sudo systemctl start zookeeper
sudo systemctl enable zookeeper

Verifica que ZooKeeper está ejecutándose:

echo ruok | nc localhost 2181

Una respuesta de imok indica que ZooKeeper está operativo.

Uso del Modo KRaft (Método Moderno)

El modo KRaft simplifica la implementación eliminando el coordinador externo. Configura Kafka en modo KRaft editando el archivo de propiedades del servidor.

Primero, genera un ID de cluster:

CLUSTER_ID=$(kafka-storage.sh random-uuid)
echo $CLUSTER_ID

Configuración de Brokers de Kafka

Crea el directorio de configuración de Kafka:

sudo mkdir -p /etc/kafka
sudo chown kafka:kafka /etc/kafka

Para modo ZooKeeper, edita la configuración del servidor:

sudo cp /opt/kafka/config/server.properties /etc/kafka/
sudo chown kafka:kafka /etc/kafka/server.properties

Edita /etc/kafka/server.properties:

sudo nano /etc/kafka/server.properties

Parámetros de configuración clave:

# Identificación del broker
broker.id=1
node.id=1

# Escuchadores y direcciones anunciadas
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://kafka-broker-1.example.com:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT

# Conexión de ZooKeeper
zookeeper.connect=localhost:2181/kafka

# Configuración de registro y almacenamiento
log.dir=/var/log/kafka
log.dirs=/var/kafka-logs

# Configuración de replicación predeterminada
default.replication.factor=3
min.insync.replicas=2

# Configuración de retención
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824

# Política de limpieza
log.cleanup.policy=delete

# Configuración de rendimiento
num.network.threads=8
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# Configuración del coordinador de grupo
group.initial.rebalance.delay.ms=3000

Para modo KRaft, crea una nueva configuración:

sudo nano /etc/kafka/kraft-server.properties

Añade estas configuraciones:

# Configuración de KRaft
process.roles=broker,controller
node.id=1
[email protected]:9093
controller.listener.names=CONTROLLER
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://kafka-broker-1.example.com:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
inter.broker.listener.name=PLAINTEXT

# Registro y almacenamiento
log.dir=/var/kafka-logs

# Otras configuraciones
log.retention.hours=168
num.network.threads=8
num.io.threads=8

Formatea el almacenamiento para KRaft:

sudo mkdir -p /var/kafka-logs
sudo chown kafka:kafka /var/kafka-logs
sudo -u kafka bash -c "CLUSTER_ID='$CLUSTER_ID' /opt/kafka/bin/kafka-storage.sh format -t $CLUSTER_ID -c /etc/kafka/kraft-server.properties"

Crea un archivo de servicio systemd para Kafka:

sudo tee /etc/systemd/system/kafka.service <<EOF
[Unit]
Description=Apache Kafka
Requires=network-online.target
After=network-online.target

[Service]
Type=simple
User=kafka
ExecStart=/opt/kafka/bin/kafka-server-start.sh /etc/kafka/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

Para el modo KRaft, usa -c /etc/kafka/kraft-server.properties en su lugar.

Inicia y habilita Kafka:

sudo systemctl daemon-reload
sudo systemctl start kafka
sudo systemctl enable kafka

Verifica que Kafka está ejecutándose:

sudo systemctl status kafka
ps aux | grep kafka

Creación y Gestión de Temas

Usa la herramienta de gestión de temas de Kafka para crear temas:

kafka-topics.sh --create \
  --topic orders \
  --partitions 3 \
  --replication-factor 2 \
  --bootstrap-server localhost:9092

Lista todos los temas:

kafka-topics.sh --list --bootstrap-server localhost:9092

Describe un tema específico:

kafka-topics.sh --describe \
  --topic orders \
  --bootstrap-server localhost:9092

Aumenta las particiones en un tema existente:

kafka-topics.sh --alter \
  --topic orders \
  --partitions 5 \
  --bootstrap-server localhost:9092

Elimina un tema:

kafka-topics.sh --delete \
  --topic orders \
  --bootstrap-server localhost:9092

Productores y Consumidores

Prueba la producción de mensajes con el productor de consola:

kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092

Esto abre un mensaje interactivo. Escribe mensajes y presiona Enter:

{"order_id": "12345", "customer": "John Doe", "amount": 99.99}
{"order_id": "12346", "customer": "Jane Smith", "amount": 149.50}

En otra terminal, inicia un consumidor de consola:

kafka-console-consumer.sh \
  --topic orders \
  --from-beginning \
  --bootstrap-server localhost:9092

La bandera --from-beginning lee todos los mensajes desde el inicio del tema. Omítela para ver solo nuevos mensajes.

Crea un grupo de consumidores para rastrear el consumo de mensajes:

kafka-console-consumer.sh \
  --topic orders \
  --group order-processor \
  --bootstrap-server localhost:9092

Lista grupos de consumidores:

kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

Describe un grupo de consumidores:

kafka-consumer-groups.sh --describe \
  --group order-processor \
  --bootstrap-server localhost:9092

Restablece el offset del grupo de consumidores al principio:

kafka-consumer-groups.sh --reset-offsets \
  --group order-processor \
  --topic orders \
  --to-earliest \
  --execute \
  --bootstrap-server localhost:9092

Retención de Mensajes y Limpieza

Controla cuánto tiempo Kafka retiene los mensajes modificando las configuraciones de los temas.

Establece el tiempo de retención a 7 días:

kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name orders \
  --alter \
  --add-config retention.ms=604800000

Establece el tamaño de retención a 1GB:

kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name orders \
  --alter \
  --add-config retention.bytes=1073741824

Configura la política de limpieza para compactar (mantiene el último valor por clave):

kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name user-profiles \
  --alter \
  --add-config cleanup.policy=compact

Ver configuración del tema actual:

kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name orders \
  --describe

Monitoreo de Kafka

Monitoriza las métricas del broker usando JMX (Java Management Extensions). Configura JMX para Kafka editando los scripts de inicio:

export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9999"

Comprueba el estado del broker y las métricas:

kafka-broker-api-versions.sh --bootstrap-server localhost:9092

Monitoriza las particiones sub-replicadas:

kafka-topics.sh --describe --under-replicated-partitions --bootstrap-server localhost:9092

Comprueba los metadatos del cluster:

kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/00000000000000000000.log --print

Solución de Problemas

Verifica los registros de Kafka en busca de errores:

sudo tail -f /var/log/kafka/server.log
sudo tail -f /var/kafka-logs/*.log

Verifica la conectividad del broker:

kafka-broker-api-versions.sh --bootstrap-server localhost:9092

Comprueba si ZooKeeper está conectado correctamente:

echo dump | nc localhost 2181 | grep brokers

Restablece el estado del broker (destructivo):

sudo systemctl stop kafka
sudo rm -rf /var/kafka-logs/*
sudo systemctl start kafka

Verifica la conectividad de red entre brokers:

nc -zv kafka-broker-2.example.com 9092
nc -zv kafka-broker-3.example.com 9092

Conclusión

Apache Kafka proporciona una plataforma escalable y tolerante a fallos para streaming de eventos y procesamiento de datos en tiempo real. Esta guía cubrió instalación, configuración en modo ZooKeeper y KRaft, gestión de temas, conceptos básicos productor-consumidor y políticas de retención. Para implementaciones en producción, implementa seguridad con TLS/SSL, configura autenticación, configura monitoreo con Prometheus y Grafana, establece procedimientos de copia de seguridad e implementa en múltiples nodos de broker para alta disponibilidad. Considera usar servicios Kafka gestionados para empresas que requieren soporte profesional y operaciones simplificadas.