Giám sát Kafka: consumer lag và các chỉ số cần theo dõi

1. Tổng quan

Một hệ thống Kafka có thể trông “khỏe mạnh”: broker vẫn chạy, producer vẫn gửi được message. Nhưng consumer có thể đang tụt lại phía sau hàng triệu message mà không ai biết. Chỉ số phản ánh tình trạng này là consumer lag.

Trong bài này, bạn sẽ hiểu consumer lag được tính thế nào và đo nó bằng ba cách: CLI, AdminClient trong Java, và Micrometer trong Spring Boot. Sau đó, chúng ta điểm qua các chỉ số broker, producer, consumer quan trọng nhất và cách đặt cảnh báo cho chúng.

2. Consumer lag là gì?

Mỗi partition trong Kafka là một log chỉ ghi nối tiếp. Có hai offset bạn cần quan tâm:

• Log-end offset (LEO): offset của message tiếp theo sẽ được ghi vào partition.
• Committed offset: offset mà consumer group đã commit, tức vị trí group sẽ đọc tiếp nếu restart.

Lag của một partition được tính như sau:

lag = log-end offset - committed offset

Lag của cả consumer group là tổng lag trên mọi partition mà group đó đọc.

Lag bằng 0 nghĩa là consumer đã đọc kịp. Lag dao động nhẹ là bình thường, vì message luôn đến liên tục. Lag tăng đều theo thời gian mới là dấu hiệu đáng lo. Khi đó, consumer xử lý chậm hơn tốc độ producer ghi vào.

Lưu ý: Với topic dùng transaction hoặc compaction, offset có thể có “khoảng trống” (control record, message bị xóa). Lag tính theo offset vì vậy có thể lớn hơn số message thực sự còn phải đọc một chút. Đây không phải lỗi.

3. Kiểm tra lag bằng CLI

Cách nhanh nhất là dùng script đi kèm Kafka:

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

Kết quả có dạng:

GROUP          TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID       HOST
order-service  orders  0          15230           15232           2     consumer-1-a1b2   /10.0.0.5
order-service  orders  1          14980           18410           3430  consumer-1-c3d4   /10.0.0.6
order-service  orders  2          15102           15102           0     consumer-1-e5f6   /10.0.0.7

Partition 1 đang tụt 3430 message, trong khi hai partition còn lại gần như theo kịp. Lag lệch như vậy thường do hai nguyên nhân. Một là có key “nóng” dồn vào một partition. Hai là consumer trên host 10.0.0.6 gặp vấn đề.

CLI rất hữu ích khi điều tra sự cố. Tuy vậy, bạn không thể ngồi chạy lệnh này mỗi phút. Phần tiếp theo sẽ tự động hóa việc đo lag.

4. Tính lag bằng AdminClient

Thêm dependency kafka-clients (Spring Boot đã quản lý version):

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

Trước tiên, tách phần tính toán thành một hàm thuần để dễ test:

public final class LagCalculator {

    private LagCalculator() {}

    public static Map<TopicPartition, Long> compute(Map<TopicPartition, Long> committed,
                                                    Map<TopicPartition, Long> endOffsets) {
        Map<TopicPartition, Long> lag = new HashMap<>();
        committed.forEach((tp, committedOffset) -> {
            Long end = endOffsets.get(tp);
            if (end != null) {
                lag.put(tp, Math.max(0L, end - committedOffset));
            }
        });
        return lag;
    }
}

Hàm này chỉ tính lag cho partition có cả committed offset lẫn end offset. Math.max(0L, …) có lý do riêng. Committed offset và end offset được lấy bằng hai request khác nhau, nên đôi khi committed offset “mới” hơn end offset vừa lấy. Kết quả âm khi đó vô nghĩa, nên ta đưa về 0.

Tiếp theo là lớp gọi AdminClient:

public class ConsumerLagChecker {

    private final Admin admin;

    public ConsumerLagChecker(Admin admin) {
        this.admin = admin;
    }

    public Map<TopicPartition, Long> lagOf(String groupId) throws Exception {
        Map<TopicPartition, Long> committed = admin.listConsumerGroupOffsets(groupId)
                .partitionsToOffsetAndMetadata()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .filter(e -> e.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        Map<TopicPartition, OffsetSpec> request = committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()));

        Map<TopicPartition, Long> endOffsets = admin.listOffsets(request)
                .all()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        return LagCalculator.compute(committed, endOffsets);
    }
}

Từng bước diễn ra như sau:

• listConsumerGroupOffsets trả về committed offset của group. Giá trị có thể là null nếu partition chưa từng được commit, nên ta lọc bỏ.
• listOffsets với OffsetSpec.latest() lấy log-end offset của từng partition.
• Mỗi lời gọi đều có timeout 10 giây. Nhờ vậy, thread không bị treo vô hạn khi cluster gặp sự cố.

Sử dụng:

try (Admin admin = Admin.create(Map.of(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"))) {
    new ConsumerLagChecker(admin).lagOf("order-service")
            .forEach((tp, lag) -> System.out.println(tp + " -> " + lag));
}

orders-0 -> 2
orders-1 -> 3430
orders-2 -> 0

4.1. Test cho LagCalculator

Dùng JUnit 5 và AssertJ (có sẵn trong spring-boot-starter-test):

class LagCalculatorTest {

    private final TopicPartition p0 = new TopicPartition("orders", 0);
    private final TopicPartition p1 = new TopicPartition("orders", 1);

    @Test
    void shouldComputeLagPerPartition() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 100L, p1, 50L),
                Map.of(p0, 120L, p1, 50L));

        assertThat(lag).containsEntry(p0, 20L).containsEntry(p1, 0L);
    }

    @Test
    void shouldClampNegativeLagToZero() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 130L),
                Map.of(p0, 120L));

        assertThat(lag).containsEntry(p0, 0L);
    }

    @Test
    void shouldSkipPartitionWithoutEndOffset() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 10L), Map.of());

        assertThat(lag).isEmpty();
    }
}

5. Đưa lag lên Micrometer trong Spring Boot

Thêm Actuator và Prometheus registry:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

management.endpoints.web.exposure.include=health,prometheus

5.1. Metric phía client có sẵn

Khi có MeterRegistry, Spring Boot tự gắn MicrometerConsumerListener vào ConsumerFactory. Mọi metric nội bộ của Kafka consumer sẽ xuất hiện ở /actuator/prometheus, trong đó có:

kafka_consumer_fetch_manager_records_lag_max{client_id="consumer-order-service-1",...} 3430.0
kafka_consumer_fetch_manager_records_lag{topic="orders",partition="1",...} 3430.0

records-lag-max là lag lớn nhất mà consumer này quan sát được. Kafka client tự tính nó từ vị trí fetch, không phải từ committed offset.

5.2. Gauge lag tự đo từ bên ngoài

Metric phía client có một điểm yếu lớn. Khi consumer chết, metric cũng biến mất. Đúng lúc lag tăng nhanh nhất thì bạn lại không còn số liệu. Vì vậy, bạn nên đo thêm lag từ một tiến trình độc lập bằng AdminClient:

@Component
public class ConsumerLagMetrics {

    private final ConsumerLagChecker checker;
    private final MeterRegistry registry;
    private final Map<TopicPartition, AtomicLong> gauges = new ConcurrentHashMap<>();

    public ConsumerLagMetrics(KafkaAdmin kafkaAdmin, MeterRegistry registry) {
        this.checker = new ConsumerLagChecker(Admin.create(kafkaAdmin.getConfigurationProperties()));
        this.registry = registry;
    }

    @Scheduled(fixedDelay = 30_000)
    public void refresh() throws Exception {
        checker.lagOf("order-service").forEach((tp, lag) ->
                gauges.computeIfAbsent(tp, this::register).set(lag));
    }

    private AtomicLong register(TopicPartition tp) {
        AtomicLong holder = new AtomicLong();
        Gauge.builder("kafka.consumer.group.lag", holder, AtomicLong::get)
                .tag("group", "order-service")
                .tag("topic", tp.topic())
                .tag("partition", String.valueOf(tp.partition()))
                .register(registry);
        return holder;
    }
}

Nhớ bật scheduling bằng @EnableScheduling trên một lớp @Configuration. Cứ mỗi 30 giây, gauge kafka_consumer_group_lag được cập nhật. Giá trị này dựa trên committed offset, nên vẫn đúng kể cả khi mọi consumer của group đã dừng.

Trong production, nhiều đội dùng công cụ chuyên dụng như kafka-exporter hoặc Burrow thay cho đoạn code trên. Nguyên lý của chúng vẫn giống hệt: so committed offset với log-end offset.

Lưu ý: Đừng đặt gauge này trong chính service consumer nếu service đó có nhiều instance. Mỗi instance sẽ xuất ra cùng một giá trị lag, và dashboard cộng dồn sẽ sai. Hãy chạy nó ở một service giám sát riêng, hoặc chỉ ở một instance.

6. Các chỉ số cần theo dõi

Lag cho biết có vấn đề. Các chỉ số dưới đây giúp bạn tìm ra vấn đề ở đâu.

6.1. Broker

Chỉ số (JMX)	Ý nghĩa	Giá trị mong muốn
UnderReplicatedPartitions	Partition có replica chưa đồng bộ	0
OfflinePartitionsCount	Partition không có leader	0
ActiveControllerCount	Số controller đang hoạt động (tổng cluster)	đúng 1
IsrShrinksPerSec / IsrExpandsPerSec	ISR co giãn	gần 0, không dao động liên tục
RequestHandlerAvgIdlePercent	Mức rảnh của request handler thread	> 0.3
NetworkProcessorAvgIdlePercent	Mức rảnh của network thread	> 0.3

OfflinePartitionsCount > 0 là sự cố nghiêm trọng: partition đó không đọc và không ghi được. UnderReplicatedPartitions kéo dài thường báo hiệu broker quá tải, đĩa chậm hoặc mạng có vấn đề.

6.2. Producer

• record-error-rate: tỷ lệ gửi thất bại sau khi đã hết retry. Mọi giá trị khác 0 đều đáng điều tra.
• request-latency-avg: độ trễ request tới broker. Nếu tăng, bạn sẽ thấy throughput giảm.
• record-retry-rate: retry nhiều nghĩa là broker hoặc mạng không ổn định.
• buffer-available-bytes: buffer cạn khiến send() bị block.

6.3. Consumer

• records-lag-max: đã nói ở trên.
• records-consumed-rate: tốc độ tiêu thụ. Hãy so với tốc độ producer ghi vào topic.
• rebalance-rate-per-hour và last-rebalance-seconds-ago: rebalance liên tục làm consumer “đứng hình”. Đây là thủ phạm phổ biến khiến lag tăng.
• commit-latency-avg: commit chậm làm giảm throughput, nhất là khi commit đồng bộ.
• last-poll-seconds-ago: nếu vượt max.poll.interval.ms, consumer bị đá khỏi group.

7. Đặt cảnh báo hợp lý

Lỗi phổ biến nhất là đặt ngưỡng tuyệt đối kiểu “lag > 1000 thì báo”. Với topic 50.000 message/giây, 1000 là con số vô nghĩa. Với topic vài message/phút, 1000 có thể đã là chậm hàng giờ.

Có hai cách tiếp cận tốt hơn. Cách thứ nhất là cảnh báo khi lag tăng liên tục:

groups:
  - name: kafka-consumer
    rules:
      - alert: KafkaConsumerLagGrowing
        expr: sum by (group, topic) (deriv(kafka_consumer_group_lag[10m])) > 0
          and sum by (group, topic) (kafka_consumer_group_lag) > 500
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Lag của {{ $labels.group }} trên {{ $labels.topic }} tăng liên tục 15 phút"

Rule này chỉ kích hoạt khi hai điều kiện cùng đúng: lag có xu hướng tăng (deriv > 0) và đã vượt mức tối thiểu. Điều kiện phải kéo dài 15 phút. Nhờ vậy, bạn bỏ qua được những đợt tăng ngắn sau khi deploy.

Cách thứ hai là quy đổi lag ra thời gian. Hãy chia lag cho tốc độ tiêu thụ để ước lượng consumer đang chậm bao nhiêu giây. Con số này gắn trực tiếp với SLA. Ví dụ: “đơn hàng phải được xử lý trong 2 phút” dễ hiểu hơn nhiều so với “lag dưới 5000”.

Ngoài ra, bạn nên có cảnh báo riêng cho trường hợp metric biến mất, chẳng hạn dùng absent(kafka_consumer_group_lag). Không có dữ liệu không có nghĩa là mọi thứ ổn.

8. Lỗi hay gặp

• Chỉ dựa vào records-lag-max. Khi consumer chết, metric này biến mất. Hãy luôn có một nguồn đo lag từ bên ngoài.
• Đọc lag của group không có consumer active. CLI vẫn hiển thị committed offset, nhưng cột CONSUMER-ID trống. Lag tăng ở đây có nghĩa là không ai đang xử lý.
• Tăng số consumer vượt số partition. Consumer thừa sẽ ngồi không, lag không giảm. Muốn tăng song song, bạn phải tăng số partition.
• Bỏ qua rebalance. Xử lý một batch quá lâu khiến vượt max.poll.interval.ms. Consumer bị loại, group rebalance, rồi lặp lại. Lag tăng theo kiểu răng cưa. Hãy giảm max.poll.records hoặc đẩy xử lý nặng sang thread khác.
• Gọi AdminClient quá dày. Truy vấn offset mỗi giây cho hàng trăm group sẽ tạo tải không cần thiết lên broker. Chu kỳ 15–60 giây là đủ cho hầu hết hệ thống.

9. Kết luận

Consumer lag là chỉ số trực tiếp nhất cho biết hệ thống Kafka có theo kịp dữ liệu hay không. Lag được tính bằng log-end offset trừ committed offset. Bạn có thể xem lag bằng kafka-consumer-groups.sh, tính bằng AdminClient hoặc xuất ra Prometheus qua Micrometer.

Để giám sát hiệu quả, hãy kết hợp metric phía client với một nguồn đo độc lập. Song song với lag, theo dõi các chỉ số broker như UnderReplicatedPartitions và OfflinePartitionsCount, cùng tần suất rebalance ở phía consumer. Cuối cùng, cảnh báo theo xu hướng và theo thời gian trễ thay vì theo một con số lag cố định. Như vậy bạn sẽ ít bị báo động giả mà không bỏ lỡ sự cố thật.