Consumer group và rebalance trong Kafka với Spring Boot

1. Tổng quan

Khi chạy nhiều instance của cùng một service đọc từ Kafka, bạn cần biết message nào do instance nào xử lý. Kafka giải quyết việc này bằng consumer group và cơ chế rebalance.

Bài này giải thích cách Kafka chia partition trong một group và những tình huống gây ra rebalance. Sau đó bạn sẽ cấu hình Spring Boot để rebalance ít xảy ra hơn và không làm mất hoặc trùng dữ liệu. Cuối bài có một test với Embedded Kafka để kiểm chứng cách chia partition.

2. Consumer group hoạt động thế nào

Một consumer group là tập hợp các consumer có cùng group.id. Kafka đảm bảo:

• Mỗi partition của topic được gán cho đúng một consumer trong group tại một thời điểm.
• Một consumer có thể nhận nhiều partition.
• Các group khác nhau đọc độc lập. Mỗi group có offset riêng.

Giả sử topic orders có 4 partition:

Group "billing" với 2 consumer:
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Group "billing" với 5 consumer:
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> (rảnh, không có partition)

Hệ quả quan trọng là số partition giới hạn độ song song tối đa của một group. Thêm consumer vượt quá số partition không giúp xử lý nhanh hơn.

Việc điều phối do một broker đảm nhiệm, gọi là group coordinator. Consumer gửi heartbeat đều đặn tới coordinator để báo rằng nó còn sống.

2.1. Dependency

Thêm Spring for Apache Kafka. Spring Boot 3.x quản lý sẵn version:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

2.2. Một listener đơn giản

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(topics = "orders", groupId = "billing", concurrency = "2")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("partition={} offset={} order={}",
                record.partition(), record.offset(), record.value());
    }
}

• groupId = "billing" đặt consumer vào group billing.
• concurrency = "2" tạo 2 consumer (2 thread) trong cùng instance. Nếu bạn chạy 2 instance của service, group sẽ có 4 consumer, mỗi consumer nhận 1 partition.

Log khi khởi động cho thấy kết quả phân chia:

billing: partitions assigned: [orders-0, orders-1]
billing: partitions assigned: [orders-2, orders-3]

3. Rebalance là gì và khi nào xảy ra

Rebalance là quá trình group phân chia lại partition giữa các thành viên. Nó xảy ra khi:

1. Consumer mới tham gia, ví dụ bạn scale thêm pod hoặc deploy instance mới.
2. Consumer rời đi một cách bình thường, ví dụ shutdown gọi close().
3. Consumer bị coi là chết: coordinator không nhận heartbeat trong session.timeout.ms.
4. Consumer xử lý quá chậm: khoảng cách giữa hai lần poll() vượt max.poll.interval.ms. Consumer tự rời group.
5. Topic thay đổi: tăng số partition, hoặc có topic mới khớp với pattern đã subscribe.

Vấn đề là trong giao thức eager cũ, mọi consumer phải trả lại toàn bộ partition rồi mới được gán lại. Cả group dừng xử lý trong lúc đó, hiện tượng này thường gọi là “stop-the-world”. Với group lớn hoặc deploy kiểu rolling, rebalance có thể lặp lại liên tục.

4. Các tham số timeout quan trọng

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: billing
      enable-auto-commit: false
      max-poll-records: 100
      properties:
        session.timeout.ms: 45000
        heartbeat.interval.ms: 3000
        max.poll.interval.ms: 300000

Tham số	Ý nghĩa
session.timeout.ms	Thời gian tối đa không có heartbeat trước khi consumer bị loại
heartbeat.interval.ms	Chu kỳ gửi heartbeat, thường ≤ 1/3 session.timeout.ms
max.poll.interval.ms	Khoảng tối đa giữa hai lần poll()
max-poll-records	Số record tối đa mỗi lần poll()

Heartbeat được gửi từ một thread nền riêng. Vì vậy consumer vẫn “sống” theo session.timeout.ms ngay cả khi thread chính đang bận. Tuy nhiên max.poll.interval.ms chỉ đo thread xử lý. Nếu một batch xử lý quá lâu, consumer vẫn bị đá khỏi group.

Lỗi điển hình trong log trông như sau:

Member consumer-billing-1 sending LeaveGroup request to coordinator due to
consumer poll timeout has expired. This means the time between subsequent
calls to poll() was longer than the configured max.poll.interval.ms ...
CommitFailedException: Commit cannot be completed since the group has already rebalanced

Công thức cần thỏa mãn:

max-poll-records × thời gian xử lý 1 record  <  max.poll.interval.ms

Lưu ý: Đừng “chữa” bằng cách tăng max.poll.interval.ms lên vài giờ. Consumer bị treo thật sẽ giữ partition rất lâu mà không ai xử lý. Hãy giảm max-poll-records, hoặc đưa tác vụ nặng ra khỏi luồng consume.

5. Giảm tác động của rebalance

5.1. Cooperative rebalancing

Với CooperativeStickyAssignor, chỉ những partition thực sự cần chuyển mới bị thu hồi. Các consumer còn lại tiếp tục xử lý partition của mình trong lúc rebalance diễn ra.

spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Ví dụ: group có A (orders-0..3), sau đó B tham gia.

Eager:       A trả hết 4 partition -> chia lại: A {0,1}, B {2,3}
Cooperative: A chỉ trả {2,3}, vẫn xử lý {0,1} liên tục -> B nhận {2,3}

Lưu ý: Không được chuyển thẳng một group đang chạy từ eager sang cooperative trong một lần deploy. Bạn cần rolling deploy hai lần. Lần đầu khai báo cả hai assignor, ví dụ CooperativeStickyAssignor,RangeAssignor. Lần sau mới bỏ RangeAssignor.

Từ Kafka 4.0, giao thức rebalance mới (KIP-848) đã GA. Bạn bật nó bằng group.protocol=consumer. Khi đó việc phân chia do broker thực hiện, rebalance mặc định đã có tính tăng dần, và các tham số partition.assignment.strategy cùng session.timeout.ms phía client không còn được dùng. Nếu cluster của bạn hỗ trợ, đây là lựa chọn nên cân nhắc đầu tiên.

5.2. Static membership

Mỗi lần pod restart, consumer nhận một member.id mới, và việc đó kích hoạt rebalance hai lần (rời rồi vào lại). Static membership gán một định danh cố định cho consumer:

spring:
  kafka:
    consumer:
      properties:
        group.instance.id: billing-${HOSTNAME}
        session.timeout.ms: 60000

Nếu consumer quay lại trong thời gian session.timeout.ms, coordinator giao lại đúng các partition cũ mà không cần rebalance. Cách này rất hợp với Kubernetes StatefulSet, nơi HOSTNAME ổn định (billing-0, billing-1…).

Lưu ý: group.instance.id phải duy nhất trong group. Với concurrency > 1, Spring Kafka tự thêm hậu tố -n cho từng consumer con. Nếu hai instance dùng cùng ID, instance vào sau sẽ “đá” instance trước ra (FencedInstanceIdException). Đừng dùng static membership với Deployment có tên pod ngẫu nhiên, vì bạn sẽ không được lợi gì.

6. Xử lý offset khi rebalance

Rủi ro lớn nhất của rebalance là xử lý trùng. Consumer A đã xử lý xong record nhưng chưa commit, partition bị chuyển sang B, và B đọc lại từ offset đã commit gần nhất.

Spring Kafka mặc định commit sau mỗi batch (AckMode.BATCH) và tự commit các offset đang chờ trước khi partition bị thu hồi. Nếu cần can thiệp, ví dụ flush buffer hoặc ghi log, bạn dùng ConsumerAwareRebalanceListener:

@Component
public class OrderRebalanceListener implements ConsumerAwareRebalanceListener {

    private static final Logger log = LoggerFactory.getLogger(OrderRebalanceListener.class);

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
                                                Collection<TopicPartition> partitions) {
        log.info("Sắp mất partition {}, flush dữ liệu đang giữ trong bộ nhớ", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("Đã commit và trả partition {}", partitions);
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        log.info("Nhận partition {}", partitions);
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("Mất partition {} mà không kịp commit", partitions);
    }
}

• onPartitionsRevokedBeforeCommit chạy trước khi container commit offset. Đây là chỗ để hoàn tất công việc dở dang.
• onPartitionsRevokedAfterCommit chạy sau khi offset đã được commit.
• onPartitionsLost được gọi khi consumer bị loại khỏi group, ví dụ vì vượt max.poll.interval.ms. Lúc này partition có thể đã thuộc consumer khác, nên đừng commit.

Đăng ký listener vào container factory:

@Configuration
public class KafkaConfig {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory,
            OrderRebalanceListener rebalanceListener) {

        var factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
        configurer.configure(factory, consumerFactory);
        factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
        return factory;
    }
}

configurer.configure(...) giữ nguyên toàn bộ cấu hình từ application.yml. Dòng tiếp theo chỉ bổ sung rebalance listener.

Lưu ý: Dù cấu hình tốt đến đâu, Kafka chỉ đảm bảo at-least-once cho consumer thông thường. Hãy làm cho xử lý của bạn idempotent, ví dụ kiểm tra orderId đã thanh toán trước khi gọi PaymentService. Đừng giả định rằng rebalance sẽ không bao giờ gây trùng.

7. Kiểm thử việc phân chia partition

Test sau dùng @EmbeddedKafka để kiểm chứng rằng hai consumer cùng group chia đều 4 partition:

@EmbeddedKafka(partitions = 4, topics = "orders")
class ConsumerGroupAssignmentTest {

    @Test
    void twoConsumersInSameGroupShareAllPartitions(EmbeddedKafkaBroker broker) {
        Map<String, Object> props = KafkaTestUtils.consumerProps("billing", "false", broker);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        try (var consumerA = new KafkaConsumer<String, String>(props);
             var consumerB = new KafkaConsumer<String, String>(props)) {

            consumerA.subscribe(List.of("orders"));
            consumerB.subscribe(List.of("orders"));

            long deadline = System.currentTimeMillis() + 30_000;
            while (System.currentTimeMillis() < deadline
                    && !(consumerA.assignment().size() == 2 && consumerB.assignment().size() == 2)) {
                consumerA.poll(Duration.ofMillis(200));
                consumerB.poll(Duration.ofMillis(200));
            }

            Set<TopicPartition> all = new HashSet<>(consumerA.assignment());
            all.addAll(consumerB.assignment());

            assertThat(consumerA.assignment()).hasSize(2);
            assertThat(consumerB.assignment()).hasSize(2)
                    .doesNotContainAnyElementsOf(consumerA.assignment());
            assertThat(all).hasSize(4);
        }
    }
}

Giải thích:

• @EmbeddedKafka khởi động một broker trong bộ nhớ, tạo topic orders với 4 partition. Broker được inject trực tiếp làm tham số của method test.
• Hai KafkaConsumer dùng chung group.id = "billing".
• Vòng lặp gọi poll() xen kẽ trên cả hai consumer, vì consumer chỉ tham gia rebalance bên trong poll(). Nếu chỉ consumer A poll, rebalance sẽ không hoàn tất. Hai consumer chạy trên cùng một thread nhưng là hai object riêng, nên không vi phạm quy tắc “một consumer chỉ được dùng bởi một thread”.
• Các assertion kiểm tra mỗi consumer có 2 partition, không trùng nhau, và tổng cộng phủ đủ 4 partition.

Đây cũng là cách tốt để bạn thử nghiệm partition.assignment.strategy khác nhau. Hãy thêm consumer thứ ba, đóng một consumer, rồi quan sát assignment() thay đổi.

8. Những lỗi hay gặp

• Số consumer lớn hơn số partition: consumer thừa sẽ ngồi không. Tăng concurrency không làm hệ thống nhanh hơn.
• Xử lý đồng bộ quá lâu trong listener: vượt max.poll.interval.ms gây vòng lặp rebalance. Consumer bị đá, quay lại, nhận lại cùng batch, rồi lại bị đá.
• Commit thủ công trong onPartitionsLost: gây CommitFailedException, hoặc tệ hơn là ghi đè offset của consumer mới.
• Dùng chung group.id cho hai service khác nhau: hai service sẽ “chia nhau” message thay vì mỗi bên nhận đủ. Mỗi service độc lập nên có group riêng.
• Rolling deploy với eager protocol trên group lớn: mỗi pod restart gây hai lần stop-the-world. Hãy dùng cooperative, static membership hoặc giao thức mới của Kafka 4.x.

9. Kết luận

Consumer group cho phép bạn scale việc đọc Kafka theo chiều ngang, với giới hạn trên là số partition. Rebalance là cái giá phải trả cho tính linh hoạt đó. Nó xảy ra khi thành viên vào, ra, chết hoặc xử lý quá chậm.

Để rebalance ít gây hại, bạn nên làm những việc sau:

• Cấu hình hợp lý max.poll.interval.ms và max-poll-records.
• Dùng CooperativeStickyAssignor, hoặc group.protocol=consumer trên Kafka 4.x.
• Bật static membership khi hạ tầng cho phép.
• Xử lý vòng đời partition qua ConsumerAwareRebalanceListener.

Cuối cùng, hãy luôn thiết kế phần xử lý idempotent, vì at-least-once là đảm bảo mặc định mà bạn cần chấp nhận.