Apache Kafka: instalar con Docker y entender los fundamentos

1# Crear carpeta de trabajo
2# Windows (PowerShell):
3mkdir kafka-lab; cd kafka-lab
4
5# Mac (Terminal):
6mkdir kafka-lab && cd kafka-lab

Crear el directorio del laboratorio Kafka

1# docker-compose.yml para Kafka con KRaft (sin ZooKeeper)
2cat > docker-compose.yml << 'EOF'
3services:
4  kafka:
5    image: confluentinc/cp-kafka:7.6.0
6    hostname: kafka
7    container_name: kafka-broker
8    ports:
9      - "9092:9092"
10    environment:
11      KAFKA_NODE_ID: 1
12      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
13      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
14      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092
15      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
16      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
17      KAFKA_PROCESS_ROLES: broker,controller
18      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
19      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
20      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
21      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
22      KAFKA_LOG_RETENTION_HOURS: 168
23      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
24    volumes:
25      - kafka-data:/var/lib/kafka/data
26volumes:
27  kafka-data:
28EOF

docker-compose.yml: Confluent Platform 7.6 (Apache Kafka 3.6) con KRaft. Un broker, perfecto para desarrollo.

1# Levantar Kafka
2# Windows (PowerShell) y Mac (Terminal):
3docker compose up -d
4
5# Verificar que está corriendo:
6docker compose ps
7
8# Esperar ~15 segundos a que arranque completamente
9# Verificar que responde:
10docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 --list
11
12# Si no hay error, Kafka está listo.

Levantar y verificar el cluster Kafka

1# Crear un topic con 3 particiones
2docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 \
3  --create --topic pedidos --partitions 3 --replication-factor 1
4
5# Verificar que se creó:
6docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 \
7  --describe --topic pedidos
8
9# Salida esperada:
10# Topic: pedidos  Partitions: 3  ReplicationFactor: 1
11# Partition: 0    Leader: 1    Replicas: 1    Isr: 1
12# Partition: 1    Leader: 1    Replicas: 1    Isr: 1
13# Partition: 2    Leader: 1    Replicas: 1    Isr: 1

Crear un topic "pedidos" con 3 particiones

1# Terminal 1 — Consumidor (escucha desde el principio):
2docker exec -it kafka-broker kafka-console-consumer \
3  --bootstrap-server localhost:9092 \
4  --topic pedidos \
5  --from-beginning
6
7# Terminal 2 — Productor (escribe mensajes):
8docker exec -it kafka-broker kafka-console-producer \
9  --bootstrap-server localhost:9092 \
10  --topic pedidos
11
12# Escribe mensajes y pulsa Enter:
13# > {"pedido_id": "ORD-001", "total": 49.99}
14# > {"pedido_id": "ORD-002", "total": 129.00}
15# > {"pedido_id": "ORD-003", "total": 15.50}
16# Verás los mensajes aparecer en la Terminal 1 instantáneamente.

Producir y consumir desde la CLI: la forma más rápida de verificar que funciona

1# Instalar la librería kafka-python
2# Windows (PowerShell) y Mac (Terminal):
3pip install kafka-python

Instalar kafka-python en tu entorno virtual

1from kafka import KafkaProducer
2import json
3import time
4
5# Crear productor
6producer = KafkaProducer(
7    bootstrap_servers=["localhost:9092"],
8    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
9    key_serializer=lambda k: k.encode("utf-8") if k else None,
10)
11
12# Producir 10 mensajes
13for i in range(10):
14    pedido = {
15        "pedido_id": f"ORD-{1000 + i}",
16        "cliente_id": f"CLI-{i % 5}",
17        "total": round(20 + i * 15.5, 2),
18        "timestamp": time.time(),
19    }
20    # La KEY determina a qué partición va el mensaje
21    future = producer.send(
22        topic="pedidos",
23        key=pedido["cliente_id"],  # Misma key = misma partición
24        value=pedido,
25    )
26    metadata = future.get(timeout=10)
27    print(f"Enviado: {pedido['pedido_id']} -> partition={metadata.partition} offset={metadata.offset}")
28
29producer.flush()
30producer.close()
31print("\nProductor finalizado. 10 mensajes enviados.")

Productor Python: la KEY decide en qué partición aterriza cada mensaje

1from kafka import KafkaConsumer
2import json
3
4consumer = KafkaConsumer(
5    "pedidos",
6    bootstrap_servers=["localhost:9092"],
7    auto_offset_reset="earliest",  # Leer desde el principio
8    group_id="mi-primer-grupo",
9    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
10)
11
12print("Consumidor escuchando topic 'pedidos'...")
13count = 0
14for message in consumer:
15    pedido = message.value
16    print(f"  [{message.partition}:{message.offset}] {pedido['pedido_id']} - {pedido['total']}EUR")
17    count += 1
18    if count >= 10:
19        break
20
21consumer.close()
22print(f"\nConsumidos {count} mensajes.")

Consumidor Python: lee mensajes indicando partición y offset de cada uno