Kafka consumer group và rebalance: eager, cooperative và KIP-848 Trung cấp

Vấn đề

Bạn thêm instance cho service đọc topic orders để tăng tốc, nhưng lag tăng vọt trong lúc rebalance trước khi giảm. Nếu autoscaler liên tục thêm rồi bớt consumer, group gần như luôn ở trạng thái rebalance và throughput thấp hơn trước khi scale.

Rebalance là hoạt động bình thường: group chia lại partition khi thành viên vào, ra, crash, hoặc số partition đổi. Vấn đề vận hành là tần suất và thời gian của nó.

Tóm tắt giao thức

	Classic eager	Classic cooperative	KIP-848 (consumer)
Kích hoạt	Mặc định ở nhiều client (kể cả KafkaJS)	Assignor CooperativeStickyAssignor (từ Kafka 2.4)	group.protocol=consumer (opt-in, GA từ 4.0)
Khi có thay đổi	Mọi consumer nhả hết partition	Chỉ thu hồi partition cần chuyển, hai vòng	Incremental qua heartbeat, không barrier toàn group
Ai tính assignment	Group leader (client)	Group leader (client)	Broker (server-side assignor)
Gián đoạn	Cả group	Chỉ phần đổi chủ	Chỉ phần đổi chủ

Java consumer mặc định partition.assignment.strategy là RangeAssignor, CooperativeStickyAssignor, và group.protocol mặc định là classic (theo docs consumer configs 4.2).

Consumer B tham gia: ba cách xử lý

◀ Trước▶ PhátTiếp ▶↺ Từ đầuBước 1/4

Consumer B tham gia group đang chạy ổn định. Ba hàng so sánh ba cách xử lý.

Consumer group: quy tắc

• Mỗi partition trong một group chỉ có một consumer tại một thời điểm; số consumer hữu ích tối đa bằng số partition.
• Consumer thừa vẫn là thành viên nhưng không có việc; thêm consumer không chia được partition nóng.
• Group khác đọc cùng topic với assignment và offset riêng.
• Key, offset và commit: xem message key và commit offset. Sau rebalance, consumer mới tiếp tục từ offset đã commit của group; việc xử lý xong nhưng chưa commit trước khi chuyển giao sẽ bị giao lại, nên side effect phải idempotent.

Quan sát eager rebalance bằng KafkaJS

Demo (examples/src/kafka-rebalance/) dùng KafkaJS (eager) trên broker Kafka 3.6 của compose repo, topic orders 3 partition, group billing-service.

Rebalance khi một consumer mới tham gia group

Phóng to sơ đồ

Partition 0 · Partition 1 · Partition 2 · Consumer A

◀ Trước▶ PhátTiếp ▶↺ Từ đầuBước 1/6

Consumer A khởi động một mình trong group "billing-service" → được gán cả 3 partition.

Chỉ có Consumer A:

[consumer-A] GROUP_JOIN — memberId=kafkajs-1a6a...4910, được gán partition: [0, 1, 2]

Consumer B tham gia cùng group:

[consumer-A] GROUP_JOIN — memberId=kafkajs-1a6a...4910, được gán partition: [2]
[consumer-B] GROUP_JOIN — memberId=kafkajs-0a70...975c, được gán partition: [0, 1]

Trạng thái	P0	P1	P2
Chỉ có A	A	A	A
Đang eager rebalance	thu hồi	thu hồi	thu hồi
Sau khi B vào	B	B	A

Trong lúc rebalance, message vẫn nằm trong Kafka; lag tăng vì producer vẫn ghi. Log GROUP_JOIN không cho biết thời gian gián đoạn; cần đo lag và khoảng cách giữa hai lần xử lý.

KIP-848: giao thức mới (Kafka 4.0+)

Theo docs Kafka 4.2, giao thức này là GA từ 4.0, server bật mặc định, nhưng client phải chọn:

• Bật bằng group.protocol=consumer ở Java consumer; không phải mặc định.
• Broker tính assignment (group.consumer.assignors, mặc định uniform và range; client chọn bằng group.remote.assignor).
• Heartbeat và session do server điều khiển: group.consumer.heartbeat.interval.ms=5000, group.consumer.session.timeout.ms=45000.
• Các config session.timeout.ms, heartbeat.interval.ms, partition.assignment.strategy không dùng được khi đã chọn consumer.
• Hạn chế được docs nêu: chưa hỗ trợ client-side assignor, rack-aware chưa đầy đủ.
• Migration online: group chuyển từ classic sang consumer khi consumer đầu tiên dùng giao thức mới tham gia (chỉ khi assignor classic không nhúng metadata tuỳ biến).

Khuyến nghị: dịch vụ Java trên Kafka 4.x nên thử group.protocol=consumer ở staging; client không phải Java (ví dụ KafkaJS) cần kiểm tra thư viện có hỗ trợ không trước khi lên kế hoạch.

Cheat-sheet config (classic protocol, Java client)

Config	Mặc định	Ý nghĩa và khuyến nghị
session.timeout.ms	45000	Không có heartbeat quá lâu thì bị loại khỏi group
heartbeat.interval.ms	3000	Thường đặt không quá 1/3 session timeout
max.poll.interval.ms	300000	Quá hạn giữa hai lần poll() thì bị loại; handler chậm gây rebalance storm
max.poll.records	500	Giảm để mỗi vòng poll xử lý kịp trong max.poll.interval.ms
group.instance.id	null	Đặt để dùng static membership: restart nhanh trong session timeout không kích hoạt rebalance
partition.assignment.strategy	Range, CooperativeSticky	Dùng cooperative để tránh dừng cả group

KafkaJS không có max.poll.interval.ms; handler dài phải giữ heartbeat trong session timeout.

Rebalance storm: handler chậm bị đá khỏi group

1. Consumer lấy 500 record, mỗi record xử lý lâu hơn dự tính.
2. Giữa hai lần poll() vượt max.poll.interval.ms, consumer bị coi là hỏng và bị loại.
3. Partition chuyển cho consumer khác, nó cũng chậm tương tự rồi cũng bị loại.
4. Group lặp trong rebalance, lag tăng, record chưa commit bị xử lý lại nhiều lần.

Cách xử lý: giảm max.poll.records, tăng max.poll.interval.ms đúng bằng thời gian xử lý tệ nhất, tách việc nặng ra luồng riêng nhưng vẫn tiếp tục poll(), và đo thời gian xử lý mỗi batch.

Khi nào dùng / không dùng

• Cooperative hoặc KIP-848: mặc định cho group lớn hoặc autoscaling.
• Static membership: workload chạy trên Kubernetes rolling update, nơi pod restart nhanh.
• Eager: chỉ khi client không hỗ trợ lựa chọn khác; chấp nhận khoảng dừng toàn group.
• Share group (bài riêng): khi cần nhiều worker hơn số partition cho job độc lập.

Ví dụ thực tế

Thêm consumer mà lag không giảm

Luồng thanh toán key theo merchantId, topic 3 partition (giả định minh hoạ), 2 consumer. Đội tăng lên 4 consumer: consumer thứ tư không có partition nào, và một merchant lớn vẫn dồn vào một partition nên lag của partition đó không đổi. Việc cần làm trước khi tăng instance là đo lag theo từng partition, rồi mới quyết định tăng partition hay đổi key.

Trade-offs

Ưu điểm

• Chia tải bằng cách thêm consumer khi còn partition.
• Giữ thứ tự record trong partition.
• Consumer chết thì group tự chuyển partition cho consumer còn lại.

Nhược điểm

• Eager rebalance dừng cả group; cần đo và đưa vào latency budget.
• Số consumer hữu ích bị giới hạn bởi số partition.
• Handler chậm vượt max.poll.interval.ms gây rebalance storm.

Checklist production

1. Dùng CooperativeStickyAssignor (hoặc KIP-848 trên 4.x), tránh eager.
2. Đặt group.instance.id ổn định cho workload restart nhiều (Kubernetes).
3. max.poll.interval.ms lớn hơn thời gian xử lý batch tệ nhất; giảm max.poll.records nếu cần.
4. Commit offset trong onPartitionsRevoked (commit offset).
5. Alert theo số rebalance/giờ và lag từng partition.
6. Không autoscale consumer quá số partition.

Sai lầm thường gặp

Sai lầm thường gặp

Tăng partition mà chưa tính việc chuyển key. Mapping hash(key) mod N đổi, record cũ và mới của cùng key nằm hai partition. Xem message key.

Sai lầm thường gặp

Thêm consumer vượt số partition. Consumer thừa nhàn rỗi, không tăng throughput.

Sai lầm thường gặp

Bỏ qua max.poll.interval.ms. Handler chậm bị loại khỏi group, tạo vòng rebalance liên tục.

Tóm tắt

Tóm tắt

• Mỗi partition trong group chỉ có một consumer; số consumer hữu ích tối đa bằng số partition.
• Eager dừng cả group; cooperative và KIP-848 chỉ chuyển phần cần đổi.
• KIP-848 GA từ Kafka 4.0, opt-in bằng group.protocol=consumer, classic vẫn là mặc định.
• Rebalance storm thường do handler chậm vượt max.poll.interval.ms.

Tự kiểm tra: (1) Ba partition và bốn consumer cùng group thì mấy consumer có việc? (2) Cooperative rebalance cần mấy vòng và vì sao? (3) Khi đã đặt group.protocol=consumer, config session.timeout.ms của client còn tác dụng không?

Tham khảo

• KIP-848 consumer rebalance protocol — Apache Kafka 4.2
• Consumer Configs — Apache Kafka 4.2
• Consuming messages — KafkaJS
• Confluent — Kafka Rebalance Protocol