Kafka consumer group, rebalance và offset

Câu này dẫn thẳng vào consumer group, cơ chế giúp Kafka vừa cho nhiều service đọc cùng topic, vừa cho nhiều instance của một service chia việc.

1. Bài toán: inventory-service scale lên 3 instance

Topic orders có 6 partition. Black Friday tới, ta chạy:

inventory-service    × 3 instance   (trừ kho)
notification-service × 2 instance   (gửi email)
analytics-service    × 1 instance

Yêu cầu: mỗi event phải được inventory xử lý đúng một lần (không phải cả 3 instance cùng trừ kho), nhưng notification và analytics cũng phải nhận event đó.

2. Vấn đề

• Nếu mỗi instance tự đọc toàn bộ topic: 3 instance inventory trừ kho 3 lần.
• Instance chết: ai đọc tiếp phần của nó, và đọc từ đâu?
• Restart: đọc lại từ đầu 7 ngày event à?

3. Giải pháp: consumer group + offset

Parallelism tối đa của một group = min(số consumer, số partition)

3 partition + 2 consumer:  C1 ← P0, P1    C2 ← P2              (bình thường)
3 partition + 3 consumer:  C1 ← P0  C2 ← P1  C3 ← P2           (đẹp)
3 partition + 5 consumer:  C1 ← P0  C2 ← P1  C3 ← P2  C4, C5 idle

4. Code từng bước

Bước 1: application.yml của inventory-service:

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP:localhost:9092}
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest        # group mới chưa có offset → đọc từ đầu
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.shop.events"
    listener:
      ack-mode: record                   # commit sau mỗi record xử lý thành công

Bước 2: Listener:

@Component
@RequiredArgsConstructor
@Slf4j
public class OrderEventListener {

    private final InventoryService inventoryService;

    @KafkaListener(topics = "orders", groupId = "inventory-service", concurrency = "2")
    public void onOrderEvent(ConsumerRecord<String, OrderEvent> record) {
        log.info("partition={} offset={} orderId={}",
                record.partition(), record.offset(), record.key());
        OrderEvent event = record.value();
        switch (event.type()) {
            case "ORDER_CREATED"   -> inventoryService.reserve(event);
            case "ORDER_CANCELLED" -> inventoryService.release(event);
            default -> { }   // bỏ qua event không quan tâm
        }
    }
}

concurrency = "2": mỗi instance tạo 2 consumer thread (2 member trong group). 3 instance × 2 = 6 consumer = vừa đúng 6 partition. Đặt cao hơn số partition chỉ tạo thêm consumer idle.

notification-service dùng groupId = "notification-service", analytics dùng "analytics-service": mỗi service một group, nên ai cũng nhận đủ event.

Bước 3: Test và quan sát lag

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

TOPIC  PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
orders 0          1520            1520            0    consumer-inventory-1-...
orders 1          1498            1530            32   consumer-inventory-1-...
orders 4          980             2100            1120 consumer-inventory-3-...  ← chậm

# Tắt 1 instance → rebalance → partition của nó chuyển sang instance khác
# Replay analytics từ 3 ngày trước (group phải đang dừng):
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group analytics-service \
    --topic orders --reset-offsets --to-datetime 2026-10-11T00:00:00.000 --execute

5. Cách hoạt động

                 Topic orders: P0 P1 P2 P3 P4 P5
                        │
     ┌──────────────────┼────────────────────┐
     ▼                  ▼                    ▼
group inventory     group notification   group analytics
 inst1: P0,P1        inst1: P0,P1,P2      inst1: P0..P5
 inst2: P2,P3        inst2: P3,P4,P5
 inst3: P4,P5

Mỗi group có bảng offset riêng trong __consumer_offsets:
 (inventory, orders, P4) = 980
 (analytics, orders, P4) = 2100

Vòng đời một record:
poll() → xử lý → commit offset+1 → poll() tiếp
crash trước commit → consumer mới đọc lại từ offset đã commit

Rebalance: trong lúc chia lại partition, consumer bị ảnh hưởng tạm dừng. Client mới dùng cooperative rebalancing (chỉ thu hồi partition cần chuyển, không dừng cả group); Kafka 4.0 có thêm giao thức consumer group mới (KIP-848) do broker điều phối. Consumer xử lý quá lâu, vượt max.poll.interval.ms (mặc định 5 phút), sẽ bị coi là chết và gây rebalance.

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: một service = một group.id cố định. Số partition ≥ số consumer tối đa dự kiến.
• Không nên: đặt group.id ngẫu nhiên mỗi lần start (mất offset, đọc lại hoặc bỏ sót). Không làm việc chậm, blocking dài trong listener mà không chỉnh max.poll.interval.ms/max.poll.records.

Đúng, nếu các instance đó cùng một group. Chỉnh một điểm: số partition mặc định khi không chỉ định do broker quyết định (num.partitions, mặc định 1), không phải lúc nào cũng là 1.

Topic orders: P0
group inventory-service:
 inst1 ← P0
 inst2 ← (không có partition)   idle
 inst3 ← (không có partition)   idle

Scale lên 3 instance không tăng throughput chút nào. Nhưng nếu "nhiều service" là các service khác chức năng (inventory, notification, analytics) thì chúng ở group khác nhau, mỗi group vẫn đọc đủ P0. Vấn đề chỉ xuất hiện khi scale ngang trong cùng một group.

Vì thế số partition là quyết định thiết kế lúc tạo topic, dựa trên throughput dự kiến và số consumer tối đa, không phải config phụ.

Offset dễ nếu đã hiểu partition. Ba ý:

1. Offset là vị trí trong một partition

P0: 0:A 1:B 2:C 3:D
P1: 0:E 1:F 2:G          ← mỗi partition đánh số riêng
Định danh một record: topic + partition + offset, ví dụ orders/P1/2 = G

2. Committed offset thuộc về consumer group

Committed offset là offset của record tiếp theo cần đọc. Xử lý xong offset 12 thì commit 13.

3. Thứ tự process và commit quyết định rủi ro

PROCESS → COMMIT:   xử lý xong, crash trước commit → đọc lại → TRÙNG   (at-least-once)
COMMIT → PROCESS:   commit xong, crash trước xử lý → bỏ qua → MẤT      (at-most-once)

Mặc định nên chọn process rồi commit, và làm consumer idempotent để chịu được trùng (xem bài 8.1).

Không nên bật. Lý do:

• enable.auto.commit=true (Kafka client thuần): commit định kỳ theo auto.commit.interval.ms, không biết bạn xử lý xong chưa. Có thể commit record đang xử lý dở → crash là mất.
• Spring Kafka mặc định đã đặt enable.auto.commit=false và để container commit theo AckMode:

// ack-mode: manual_immediate
@KafkaListener(topics = "orders", groupId = "inventory-service")
public void onOrderEvent(OrderEvent event, Acknowledgment ack) {
    inventoryService.reserve(event);   // idempotent theo eventId
    ack.acknowledge();                 // chỉ commit khi đã xử lý xong
}

Listener ném exception thì không commit; DefaultErrorHandler sẽ seek lại và retry (bài 8.9).

Q1. Deploy 5 consumer nhưng topic chỉ có 2 partition thì sao?

🇻🇳 Tiếng Việt:

Nếu 5 consumer cùng một group thì chỉ tối đa 2 consumer được giao partition, 3 consumer còn lại idle, vì trong một group mỗi partition chỉ giao cho một consumer. Parallelism tối đa bằng min(số consumer, số partition). Muốn scale phải tăng partition, chú ý tăng partition làm đổi mapping key.

🇬🇧 English:

If all five consumers are in the same group, at most two get a partition and the other three stay idle, because within a group a partition is assigned to only one consumer. Maximum parallelism is min(consumers, partitions). To scale further we add partitions, keeping in mind that this changes the key-to-partition mapping.

Q2. Offset là gì, commit offset để làm gì?

🇻🇳 Tiếng Việt:

Offset là vị trí của record trong một partition. Mỗi consumer group commit offset của record tiếp theo cần đọc cho từng partition vào topic __consumer_offsets. Khi consumer restart hoặc rebalance, consumer mới tiếp tục từ offset đã commit. Commit sau khi xử lý cho at-least-once, nên consumer phải idempotent.

🇬🇧 English:

An offset is the position of a record within a partition. Each consumer group commits, per partition, the offset of the next record to read into the __consumer_offsets topic. After a restart or rebalance, the new owner resumes from the committed offset. Committing after processing gives at-least-once, so the consumer must be idempotent.

Q3. Rebalance là gì, khi nào xảy ra, ảnh hưởng gì?

🇻🇳 Tiếng Việt:

Rebalance là việc chia lại partition giữa các consumer trong group khi có consumer vào, ra, chết, vượt max.poll.interval.ms, hoặc số partition thay đổi. Trong lúc đó các partition bị chuyển tạm dừng xử lý, và record chưa commit có thể bị xử lý lại. Em giảm ảnh hưởng bằng cooperative rebalancing, xử lý nhanh trong listener và consumer idempotent.

🇬🇧 English:

A rebalance redistributes partitions among group members when a consumer joins, leaves, crashes, exceeds max.poll.interval.ms, or the partition count changes. Moved partitions pause during the rebalance, and uncommitted records may be processed again. I reduce the impact with cooperative rebalancing, fast listeners and idempotent consumers.

Q4. Consumer lag là gì, xử lý thế nào khi lag tăng?

🇻🇳 Tiếng Việt:

Lag là khoảng cách giữa offset cuối của partition và offset group đã commit. Lag tăng đều nghĩa là consumer không theo kịp. Em kiểm tra listener có chậm hoặc gọi I/O blocking không, tăng số consumer nếu còn partition trống, hoặc tăng partition, và theo dõi lag bằng metrics để cảnh báo.

🇬🇧 English:

Lag is the gap between a partition's log-end offset and the group's committed offset. Steadily growing lag means consumers cannot keep up. I check for slow or blocking work in the listener, add consumers if partitions are available, or add partitions, and I alert on lag through metrics.

Cách nhớ ngắn gọn

• Một service = một group. Khác group = mỗi group nhận đủ.
• Trong group: một partition ↔ một consumer. Song song = min(consumer, partition).
• Offset theo (group, topic, partition); commit = offset tiếp theo.
• Process rồi commit + idempotent. Không dùng auto commit.