Yêu cầu và lựa chọn setup

Cần có: Docker Desktop (hoặc Docker Engine + Compose plugin) phiên bản 24+. Không cần Java, không cần cài gì thêm.

KRaft mode (không cần ZooKeeper) là lựa chọn đúng đắn cho project mới. Setup gọn hơn, ít service hơn, ít bug hơn. Image apache/kafka hỗ trợ KRaft từ Kafka 3.7+.

docker-compose.yml

version: '3.8'

services:

  kafka:
    image: apache/kafka:3.9.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      # KRaft mode — không cần ZooKeeper
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      # Topic defaults
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
    healthcheck:
      test: ["CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--list"]
      interval: 10s
      timeout: 5s
      retries: 5

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    depends_on:
      kafka:
        condition: service_healthy

Khởi động và verify

docker compose up -d

docker compose ps
# kafka      running (healthy)
# kafka-ui   running

CLI operations: topic, produce, consume

Kafka ship kèm các script CLI trong /opt/kafka/bin/. Exec vào container để chạy:

Tạo topic

# Tạo topic với 3 partitions
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create \
  --topic order-events \
  --partitions 3 \
  --replication-factor 1

# List tất cả topic
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --list

# Chi tiết một topic
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic order-events

Produce message từ CLI

docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --property "parse.key=true" \
  --property "key.separator=:"

# Nhập message theo format key:value
> order-1:{"id":1,"status":"placed","total":150000}
> order-2:{"id":2,"status":"placed","total":280000}
> order-1:{"id":1,"status":"paid","total":150000}
# Ctrl+C để thoát

Consume message từ CLI

# Đọc từ đầu (--from-beginning)
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --from-beginning \
  --property "print.key=true" \
  --property "key.separator= | "

# Output:
# order-1 | {"id":1,"status":"placed","total":150000}
# order-2 | {"id":2,"status":"placed","total":280000}
# order-1 | {"id":1,"status":"paid","total":150000}

Python quickstart với confluent-kafka

pip install confluent-kafka

Producer

from confluent_kafka import Producer
import json

conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)

def delivery_report(err, msg):
    if err:
        print(f'Delivery failed: {err}')
    else:
        print(f'Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}')

orders = [
    {'order_id': 'ord-001', 'status': 'placed', 'amount': 150000},
    {'order_id': 'ord-002', 'status': 'placed', 'amount': 280000},
    {'order_id': 'ord-001', 'status': 'paid',   'amount': 150000},
]

for order in orders:
    producer.produce(
        topic='order-events',
        key=order['order_id'],          # key → routing đến cùng partition
        value=json.dumps(order),
        callback=delivery_report
    )

producer.flush()  # đảm bảo gửi hết trước khi thoát

Consumer

from confluent_kafka import Consumer, KafkaError
import json

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processor',       # consumer group ID
    'auto.offset.reset': 'earliest',       # đọc từ đầu nếu chưa có offset
    'enable.auto.commit': 'true',           # tự commit offset (đơn giản nhất)
}

consumer = Consumer(conf)
consumer.subscribe(['order-events'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)   # poll 1 giây
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise Exception(msg.error())

        order = json.loads(msg.value().decode('utf-8'))
        print(f"[P{msg.partition()}|O{msg.offset()}] {order['order_id']}: {order['status']}")

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

Kafka UI — explore qua browser

Kafka UI tại http://localhost:8080 cho phép bạn:

• Topics — xem list topic, số partition, offset, retention
• Messages — browse message trong partition, search theo key/value/offset
• Consumers — xem consumer groups, lag (số message chưa đọc)
• Brokers — thông tin cluster, disk usage, config
• Schema Registry — nếu bạn thêm Schema Registry vào compose

Lỗi thường gặp và cách fix

Checklist verify setup hoàn chỉnh

• docker compose ps — cả hai service đều running (healthy)
• Tạo topic order-events thành công không có lỗi
• --list hiển thị order-events
• Produce 3 message từ CLI hoặc Python
• Consume hiển thị đúng 3 message theo thứ tự trong từng partition
• Kafka UI tại localhost:8080 thấy topic và consumer group