Dữ liệu không
đứng yên.
Kafka cũng vậy.

# Cần: pip install confluent-kafka
from confluent_kafka import Producer

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

# Cùng key thường đi vào cùng một partition
producer.produce(
    topic="orders.events",
    key="order-1001",
    value='{"type":"OrderCreated","orderId":"1001"}'
)
producer.flush()  # Chờ gửi xong trước khi thoát

# Cần: pip install confluent-kafka
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "billing-service",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders.events"])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print(msg.error())
            continue
        print(msg.partition(), msg.offset(), msg.value())
        # Sau khi xử lý nghiệp vụ thành công:
        consumer.commit(message=msg, asynchronous=False)
finally:
    consumer.close()

# Chạy trên máy đã cài Kafka và khởi động broker
# 1. Tạo topic với 3 partitions
bin/kafka-topics.sh --create \
  --topic orders.events --partitions 3 \
  --replication-factor 1 --bootstrap-server localhost:9092

# 2. Mở producer console, nhập theo dạng key:value
bin/kafka-console-producer.sh \
  --topic orders.events --bootstrap-server localhost:9092 \
  --property parse.key=true --property key.separator=:
# Sau đó nhập: order-1001:{"type":"OrderCreated"}

# 3. Ở terminal khác, đọc bằng consumer group
bin/kafka-console-consumer.sh \
  --topic orders.events --bootstrap-server localhost:9092 \
  --group demo-group --from-beginning \
  --property print.key=true