Consumer group và rebalance trong Kafka: vì sao consumer bị dừng

1. Tổng quan

Bạn có một service đọc message từ Kafka và chạy ổn định nhiều ngày. Một hôm consumer lag tăng vọt. Log đầy dòng Revoking previously assigned partitions và CommitFailedException. Thủ phạm thường là rebalance của consumer group.

Bài này giải thích consumer group phân chia partition ra sao, khi nào rebalance xảy ra và vì sao nó làm consumer “đứng hình”. Đọc xong, bạn sẽ biết cách cấu hình Spring Kafka để giảm số lần rebalance và giảm thời gian mỗi lần dừng.

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

Một consumer group là tập consumer dùng chung group.id. Kafka chia các partition của topic cho các thành viên trong group. Mỗi partition chỉ được giao cho một consumer tại một thời điểm.

Giả sử topic orders có 6 partition. Bạn chạy 3 instance của order-service:

order-service-1  -> orders-0, orders-1
order-service-2  -> orders-2, orders-3
order-service-3  -> orders-4, orders-5

Nếu chạy 7 instance, sẽ có một instance ngồi không, vì không còn partition để giao.

Trước hết, thêm dependency:

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

Một listener đơn giản trong Spring Boot 3.x:

@Component
public class OrderListener {

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

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("Nhận order {} từ partition {}", record.key(), record.partition());
    }
}

Annotation @KafkaListener tạo một listener container. Bên trong container là một thread chạy vòng lặp poll(). Mọi instance dùng groupId = "order-service" sẽ tham gia cùng một group. Một broker đóng vai group coordinator quản lý danh sách thành viên và việc phân chia partition.

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

Rebalance là quá trình group coordinator phân chia lại partition cho các thành viên. Các tình huống gây rebalance:

1. Consumer mới tham gia: scale up, deploy instance mới.
2. Consumer rời group chủ động: shutdown, rolling restart.
3. Consumer bị coi là chết vì không gửi heartbeat trong session.timeout.ms. Mặc định là 45 giây.
4. Consumer không gọi poll() trong max.poll.interval.ms. Mặc định là 5 phút.
5. Số partition của topic thay đổi, hoặc subscription theo pattern khớp thêm topic mới.

Hai cơ chế timeout ở mục 3 và 4 khác nhau và hay bị nhầm:

Cấu hình	Mặc định	Kiểm tra điều gì
heartbeat.interval.ms	3000	Chu kỳ gửi heartbeat (từ thread nền)
session.timeout.ms	45000	Process còn sống không
max.poll.interval.ms	300000	Thread xử lý có bị kẹt không

Heartbeat được gửi từ một thread nền riêng. Vì vậy process vẫn “sống” khi thread xử lý đang kẹt. Ngược lại, max.poll.interval.ms phát hiện đúng trường hợp code xử lý message của bạn quá chậm.

4. Vì sao consumer bị dừng

4.1. Eager rebalance: dừng toàn bộ group

Với giao thức truyền thống và các assignor kiểu eager (RangeAssignor, RoundRobinAssignor, StickyAssignor), mỗi lần rebalance diễn ra như sau:

1. Mọi consumer thu hồi toàn bộ partition đang giữ.
2. Mọi consumer gửi lại yêu cầu JoinGroup và chờ nhau.
3. Leader của group tính phương án phân chia mới, rồi coordinator gửi kết quả về.
4. Consumer nhận partition mới và tiếp tục xử lý.

Trong lúc diễn ra các bước 1–3, không ai trong group xử lý message. Chỉ một instance mới khởi động cũng làm cả group dừng lại. Đây là hiện tượng “stop-the-world”.

4.2. Xử lý quá chậm: vượt max.poll.interval.ms

Nguyên nhân phổ biến nhất trong thực tế là code xử lý chậm. Ví dụ sau gọi một service bên ngoài mất 1 giây cho mỗi message:

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, String> record) throws InterruptedException {
    paymentService.charge(record.value()); // giả sử mất ~1 giây
}

Mặc định max.poll.records = 500. Một lần poll() có thể trả về 500 record, tức khoảng 500 giây xử lý. Con số này vượt 300 giây của max.poll.interval.ms. Khi đó consumer tự rời group, và bạn sẽ thấy log như sau:

[Consumer clientId=consumer-order-service-1, groupId=order-service] consumer poll timeout has expired.
This means the time between subsequent calls to poll() was longer than the configured
max.poll.interval.ms, which typically implies that the poll loop is spending too much time
processing messages. You can address this either by increasing max.poll.interval.ms or by
reducing the maximum size of batches returned in poll() with max.poll.records.

Sau đó lệnh commit offset thất bại:

org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed
since the consumer is not part of an active group for auto partition assignment; it is likely
that the consumer was kicked out of the group.

Hậu quả kéo theo một chuỗi sự cố:

• Partition của consumer này được giao cho người khác.
• Offset chưa commit được, nên người nhận partition xử lý lại các message đó.
• Consumer cũ join lại, gây thêm một lần rebalance nữa.
• Nếu batch tiếp theo vẫn chậm, vòng lặp lặp lại mãi. Lag tăng không ngừng.

5. Cách khắc phục

5.1. Điều chỉnh kích thước batch và thời gian poll

Quy tắc cần nhớ: max.poll.records × thời gian xử lý mỗi record phải nhỏ hơn hẳn max.poll.interval.ms.

spring:
  kafka:
    consumer:
      group-id: order-service
      max-poll-records: 50
      properties:
        max.poll.interval.ms: 120000

Với 50 record và 1 giây mỗi record, một batch mất khoảng 50 giây. Con số này nằm an toàn trong giới hạn 120 giây.

Lưu ý: Đừng tăng max.poll.interval.ms lên vài giờ để “chữa cháy”. Giá trị này cũng là thời gian tối đa để Kafka phát hiện một consumer bị treo thật sự. Ngoài ra, một rebalance phải chờ consumer chậm nhất gọi poll() lại, nên giá trị lớn làm rebalance kéo dài theo.

Nếu từng message vốn xử lý lâu, ví dụ gọi API chậm, bạn nên đặt timeout cho lời gọi ngoài. Một cách khác là đẩy message sang hàng đợi nội bộ, thay vì giữ thread poll quá lâu.

5.2. Cooperative rebalancing

CooperativeStickyAssignor dùng incremental rebalance. Consumer chỉ thu hồi những partition thực sự phải chuyển đi, còn các partition khác vẫn được xử lý bình thường.

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

Khi thêm order-service-4 vào group 3 instance ở mục 2, chỉ vài partition bị chuyển sang instance mới:

order-service-1  -> orders-0, orders-1        (giữ nguyên, không dừng)
order-service-2  -> orders-2                  (mất orders-3)
order-service-3  -> orders-4                  (mất orders-5)
order-service-4  -> orders-3, orders-5

Lưu ý: Không được đổi thẳng từ eager sang cooperative trên một group đang chạy. Bạn cần rolling deploy hai lần. Lần đầu, cấu hình cả hai: CooperativeStickyAssignor,RangeAssignor. Lần sau, bỏ RangeAssignor. Nếu các thành viên không có assignor chung, group sẽ báo lỗi.

5.3. Static membership

Mỗi lần rolling restart, một instance rời group rồi quay lại. Kết quả là hai lần rebalance cho mỗi pod. Static membership giải quyết việc này. Bạn gán cho mỗi instance một định danh cố định qua group.instance.id:

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

Consumer có group.instance.id sẽ không gửi LeaveGroup khi tắt. Nếu nó quay lại trong session.timeout.ms với cùng id, coordinator trả lại đúng các partition cũ mà không rebalance.

Lưu ý: HOSTNAME phải ổn định qua các lần restart, ví dụ tên pod của StatefulSet trên Kubernetes. Với Deployment thông thường, tên pod đổi mỗi lần, nên static membership không có tác dụng. Hai instance chạy cùng lúc với cùng id sẽ đá nhau ra khỏi group (FencedInstanceIdException). Khi concurrency > 1, Spring Kafka tự thêm hậu tố -0, -1… vào id cho từng consumer con.

Đánh đổi của cách này: khi một instance chết thật, partition của nó sẽ không được ai xử lý cho tới khi hết session.timeout.ms.

5.4. Giao thức consumer group mới

Từ Kafka 4.0, giao thức rebalance mới (KIP-848) đã sẵn sàng cho production. Bạn bật nó bằng group.protocol=consumer. Việc phân chia partition chuyển hẳn về broker và luôn diễn ra theo kiểu incremental. Consumer không còn phải chờ nhau qua một “rào chắn” JoinGroup chung.

Với giao thức này, partition.assignment.strategy phía client không còn được dùng. Thay vào đó, broker chọn assignor (cấu hình phía client là group.remote.assignor). Nếu cluster và client của bạn đủ mới, đây là lựa chọn đáng cân nhắc. Bạn cần kiểm tra phiên bản kafka-clients mà Spring Boot đang kéo về trước khi bật.

6. Xử lý khi bị thu hồi partition

Dù tối ưu tới đâu, rebalance vẫn sẽ xảy ra. Việc quan trọng là commit offset đã xử lý trước khi mất partition, để giảm xử lý trùng.

Spring Kafka đã tự commit các offset đang chờ trước khi partition bị thu hồi. Nếu bạn dùng KafkaConsumer trực tiếp, bạn phải tự làm việc này bằng ConsumerRebalanceListener:

public class CommitOnRevokeListener implements ConsumerRebalanceListener {

    private final Consumer<String, String> consumer;
    private final Map<TopicPartition, OffsetAndMetadata> pending = new ConcurrentHashMap<>();

    public CommitOnRevokeListener(Consumer<String, String> consumer) {
        this.consumer = consumer;
    }

    public void markProcessed(ConsumerRecord<String, String> record) {
        pending.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
        for (TopicPartition tp : partitions) {
            OffsetAndMetadata offset = pending.remove(tp);
            if (offset != null) {
                toCommit.put(tp, offset);
            }
        }
        if (!toCommit.isEmpty()) {
            consumer.commitSync(toCommit);
        }
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Có thể nạp lại state cục bộ cho các partition mới tại đây
    }
}

Giải thích từng phần:

• markProcessed() ghi nhận offset kế tiếp cần đọc, tức offset + 1, sau khi một record xử lý xong.
• onPartitionsRevoked() chạy trên chính thread poll, ngay trước khi partition bị lấy đi. Nó chỉ commit offset của các partition bị thu hồi. Đây là điểm quan trọng với cooperative rebalancing, vì khi đó chỉ một phần partition bị thu hồi.
• commitSync() chặn tới khi broker xác nhận, đảm bảo offset đã lưu trước khi consumer khác nhận partition.

Bạn đăng ký listener khi subscribe: consumer.subscribe(List.of("orders"), listener).

Kiểm tra hành vi này bằng MockConsumer có sẵn trong kafka-clients:

class CommitOnRevokeListenerTest {

    @Test
    void shouldCommitOnlyRevokedPartitions() {
        MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        TopicPartition p0 = new TopicPartition("orders", 0);
        TopicPartition p1 = new TopicPartition("orders", 1);
        consumer.assign(List.of(p0, p1));

        CommitOnRevokeListener listener = new CommitOnRevokeListener(consumer);
        listener.markProcessed(new ConsumerRecord<>("orders", 0, 41L, "order-1", "{}"));
        listener.markProcessed(new ConsumerRecord<>("orders", 1, 7L, "order-2", "{}"));

        listener.onPartitionsRevoked(List.of(p0));

        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(Set.of(p0, p1));
        assertThat(committed.get(p0).offset()).isEqualTo(42L);
        assertThat(committed.get(p1)).isNull();
    }
}

Test giả lập việc chỉ orders-0 bị thu hồi. Kết quả mong đợi là offset 42 được commit cho orders-0. Còn orders-1 vẫn đang được giữ, nên chưa có offset nào được commit.

Lưu ý: Code trong onPartitionsRevoked() phải nhanh. Callback này chạy trong lúc rebalance. Nếu nó chậm, cả group phải chờ theo.

7. Lỗi hay gặp

• Xử lý bất đồng bộ nhưng vẫn commit theo poll. Bạn đẩy record sang một ExecutorService, rồi container commit offset ngay sau poll(). Nếu rebalance xảy ra, message đang xử lý dở có thể bị mất. Chỉ commit offset của record đã xử lý xong.
• Nhầm session.timeout.ms với max.poll.interval.ms. Tăng session timeout không giúp gì khi nguyên nhân là xử lý chậm.
• Đặt concurrency lớn hơn số partition. Các consumer thừa không có việc. Chúng chỉ làm group đông hơn và rebalance lâu hơn.
• GC pause dài hoặc CPU bị throttle trên container. Heartbeat thread không chạy kịp, consumer bị đuổi khỏi group dù code không chậm. Hãy kiểm tra metric JVM và CPU limit.
• Không thiết kế xử lý idempotent. Rebalance luôn có thể gây xử lý trùng một số message. Đừng giả định “exactly once” nếu bạn chưa dùng transaction hoặc chưa có cơ chế khử trùng, ví dụ kiểm tra orderId đã xử lý chưa.

8. Kết luận

Rebalance là cơ chế cần thiết để consumer group tự co giãn và chịu lỗi. Cái giá là consumer phải tạm dừng. Nguyên nhân phổ biến nhất khiến consumer “bị dừng” liên tục là vòng poll() vượt quá max.poll.interval.ms. Khi đó consumer bị đuổi khỏi group, commit thất bại và rebalance lặp lại mãi.

Để giảm thiểu, bạn có thể:

• Giữ mỗi batch nhỏ bằng max.poll.records.
• Chuyển sang CooperativeStickyAssignor hoặc giao thức consumer group mới.
• Dùng static membership cho các deployment có định danh ổn định.
• Commit offset đúng lúc trong onPartitionsRevoked().

Cuối cùng, hãy luôn thiết kế xử lý message idempotent. Rebalance không thể loại bỏ hoàn toàn, chỉ có thể làm cho nó ít xảy ra hơn và ít gây hại hơn.