Consumer Group — phân chia partition

Khi nhiều consumer join cùng group.id, Kafka tự động phân chia các partition của topic cho từng consumer. Quy tắc: mỗi partition chỉ được assign cho đúng một consumer trong group.

Multiple consumer groups: Hai group khác nhau (order-processor và analytics) đọc cùng một topic hoàn toàn độc lập — mỗi group có offset riêng, không ảnh hưởng nhau. Đây là điểm mạnh của Kafka so với traditional message queue.

Offset management

Offset là vị trí consumer đã đọc đến trong mỗi partition. Kafka lưu offset trong internal topic __consumer_offsets. Consumer group commit offset để báo "đã xử lý đến đây".

Auto commit (enable.auto.commit=true)

Consumer tự động commit offset định kỳ (mặc định 5 giây). Đơn giản nhưng có thể gây mất message: commit xảy ra trước khi xử lý xong → crash → message coi như đã xử lý.

Manual commit

Bạn tự gọi consumer.commit() sau khi xử lý thành công. An toàn hơn nhưng phức tạp hơn. Có hai dạng: synchronous (commit()) và asynchronous (commit_async()).

Rebalancing

Rebalancing xảy ra khi:

• Consumer mới join group
• Consumer rời group (crash hoặc shutdown graceful)
• Topic thêm partition mới
• Consumer không gửi heartbeat kịp (session.timeout.ms hết)

Trong quá trình rebalance, toàn bộ consumer trong group dừng đọc (stop-the-world). Kafka 2.4+ có incremental cooperative rebalancing — chỉ revoke partition thực sự cần di chuyển, giảm downtime đáng kể.

Delivery Semantics

Configs quan trọng

Manual commit pattern thực chiến

from confluent_kafka import Consumer, KafkaError
import json

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processor',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': 'false',  # manual commit
}
consumer = Consumer(conf)
consumer.subscribe(['order-events'])

def process_order(order):
    # business logic — idempotent nếu có thể
    print(f"Processing {order['order_id']}: {order['status']}")

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None or msg.error():
            continue

        try:
            order = json.loads(msg.value())
            process_order(order)
            consumer.commit(msg)  # commit sau khi xử lý thành công
        except Exception as e:
            print(f"Error processing: {e}")
            # không commit → message sẽ được đọc lại sau rebalance/restart
finally:
    consumer.close()

Dead Letter Queue pattern

Nếu message không thể xử lý được (malformed, business rule fail), đừng để consumer loop mãi. Gửi message đó sang một topic khác (order-events-dlq) để xử lý riêng:

from confluent_kafka import Consumer, Producer, KafkaError
import json

MAX_RETRIES = 3

def send_to_dlq(producer, original_msg, error_reason):
    dlq_value = {
        'original_topic': original_msg.topic(),
        'original_partition': original_msg.partition(),
        'original_offset': original_msg.offset(),
        'error': str(error_reason),
        'payload': original_msg.value().decode('utf-8'),
    }
    producer.produce(
        topic='order-events-dlq',
        key=original_msg.key(),
        value=json.dumps(dlq_value)
    )
    producer.flush()