Kafka Consumer Groups thực sự hoạt động như thế nào?

Hầu hết tutorial về Kafka dừng lại ở câu “consumer group giúp scale việc đọc message”. Câu đó đúng, nhưng không giải thích được những gì bạn sẽ gặp ở production: consumer đứng im dù topic còn đầy message, message bị xử lý hai lần sau khi deploy, hay cả group “đóng băng” vài giây mỗi khi một pod khởi động lại.

Bài này đi từ mô hình cơ bản đến cơ chế rebalance — nguồn gốc của phần lớn các sự cố đó.

Mô hình cơ bản: partition là đơn vị song song

Một topic được chia thành nhiều partition. Trong một consumer group, mỗi partition được giao cho đúng một consumer tại một thời điểm, còn một consumer có thể nhận nhiều partition[1].

Hệ quả trực tiếp: số consumer làm việc thực sự không bao giờ vượt quá số partition.

Offset: consumer nhớ mình đã đọc đến đâu

Kafka không xoá message khi consumer đọc xong. Thay vào đó, mỗi group lưu lại offset — vị trí đã xử lý — cho từng partition vào một topic nội bộ tên là __consumer_offsets.

Thời điểm commit offset quyết định ngữ nghĩa giao nhận:

• Commit trước khi xử lý → nếu consumer chết giữa chừng, message bị mất (at-most-once).
• Commit sau khi xử lý → nếu consumer chết sau khi xử lý nhưng trước khi commit, message bị xử lý lại (at-least-once).

Trong thực tế, at-least-once kết hợp với xử lý idempotent là lựa chọn phổ biến nhất:

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
// Tắt auto commit: chỉ commit sau khi xử lý thành công
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("orders"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> record : records) {
            orderService.handle(record.key(), record.value()); // phải idempotent
        }
        consumer.commitSync();
    }
}

Rebalance: khi group thay đổi thành viên

Rebalance là quá trình chia lại partition cho các consumer. Nó xảy ra khi:

1. Một consumer mới tham gia group (scale up, deploy phiên bản mới).
2. Một consumer rời group — chủ động (shutdown) hoặc bị coi là đã chết.
3. Số partition của topic thay đổi, hoặc tập topic mà group subscribe thay đổi.

Một broker đóng vai trò group coordinator theo dõi thành viên của group thông qua heartbeat. Có hai ngưỡng thời gian quan trọng, rất hay bị nhầm lẫn:

Eager vs. cooperative rebalance

Với giao thức rebalance truyền thống (eager), mọi consumer phải thu hồi toàn bộ partition trước khi nhận phân bổ mới. Trong khoảng thời gian đó, cả group ngừng xử lý — hiện tượng thường được gọi là “stop-the-world”.

Giao thức incremental cooperative rebalancing (KIP-429[3]) chỉ thu hồi những partition thực sự cần chuyển chủ, các partition còn lại tiếp tục được xử lý. Với Java client, bạn bật nó bằng assignor tương ứng:

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Xa hơn nữa, KIP-848[4] chuyển logic phân bổ partition từ phía client về broker, giúp rebalance nhẹ hơn và dễ vận hành hơn. Hãy kiểm tra phiên bản Kafka và client bạn đang dùng trước khi bật giao thức mới.

Quan sát consumer group trong thực tế

Công cụ dòng lệnh đi kèm Kafka cho biết partition nào đang thuộc consumer nào và lag — số message chưa xử lý — của từng partition:

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

Tóm tắt

• Partition là đơn vị song song; số consumer hữu ích bị giới hạn bởi số partition.
• Offset được lưu theo group, và thời điểm commit quyết định at-most-once hay at-least-once.
• Rebalance là cái giá phải trả cho tính đàn hồi; hiểu các timeout và dùng cooperative rebalancing giúp giảm đáng kể thời gian gián đoạn[2].