Consumer lag trong Kafka: đo, cảnh báo và xử lý Trung cấp

Vấn đề

Dashboard báo lag = 40 cho group billing. Con số nhỏ, không ai để ý. Thực tế record cũ nhất đã chờ 90 giây vì database phía sau timeout, và khách thấy hoá đơn trễ.

Hôm khác dashboard báo lag = 50.000 lúc 9h sáng, on-call bị gọi dậy. Mười giây sau lag về 0: chỉ là producer ghi dồn một batch.

Cả hai đều là cùng một lỗi: đo lag bằng số offset rồi alert bằng ngưỡng cố định. Bài này chỉ cách đo đúng, alert đúng và chẩn đoán nguyên nhân.

🎮 Thử tay

• Kafka consumer lag tăng mãi: hot partition, rebalance storm, poison message.

Tóm tắt nhanh

Câu hỏi	Trả lời ngắn
Lag là gì?	Với mỗi partition: log-end offset trừ committed offset của group
Đơn vị?	Khoảng cách offset, không hẳn là số record (compaction, transaction marker)
Đo ở đâu?	Phía ngoài, từ committed offset (CLI, Admin API, exporter); metric client chỉ bổ sung
SLI nên dùng	Time lag: tuổi của record cũ nhất chưa commit
Alert khi	Time lag vượt SLO kéo dài, hoặc lag tăng liên tục; theo partition xấu nhất
Thêm consumer giúp khi	Còn partition chưa có consumer và bottleneck nằm ở CPU của consumer

Lag được tính thế nào

Mỗi partition có ba vị trí đáng quan tâm:

Vị trí	Ý nghĩa
High watermark (HW)	Offset kế tiếp sau record cuối đã được replicate đủ; consumer read_uncommitted đọc được tới đây
Last stable offset (LSO)	Với read_committed: dừng trước transaction còn mở đầu tiên (LSO)
Committed offset	Offset group sẽ đọc tiếp, do consumer commit (commit offset)

lag = end offset − committed offset, tính từng partition. Lag của cả group là tổng, nhưng tổng che mất partition xấu nhất.

Log-end offset và committed offset theo thời gian (số liệu minh hoạ)

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

Lag của một partition = log-end offset trừ committed offset. Ban đầu consumer theo kịp: 0 offset, 0 giây.

Hai hình dạng cần nhớ từ sơ đồ:

• Số offset lớn, tuổi nhỏ: burst đang được xả. Thường vô hại.
• Số offset nhỏ, tuổi lớn: consumer kẹt trên partition ít traffic. Đây là sự cố.

Đọc lag bằng kafka-consumer-groups.sh

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group my-group

Theo mã nguồn ConsumerGroupCommand (nhánh 4.1), header in ra là:

GROUP  TOPIC  PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID  HOST  CLIENT-ID

Ví dụ trong docs Kafka 4.2 bỏ cột GROUP; output thật có cột này. Thêm --verbose thì có cột LEADER-EPOCH.

Cột	Đọc thế nào
CURRENT-OFFSET	Committed offset của group cho partition đó
LOG-END-OFFSET	Tool hỏi offset latest qua Admin API với isolation mặc định read_uncommitted, tức là high watermark
LAG	Hiệu hai cột trên
CONSUMER-ID = -	Partition không có member nào đang giữ: group trống hoặc đang rebalance
CURRENT-OFFSET = -	Group chưa commit gì cho partition này

Các option hữu ích khác (theo docs 4.2):

• --members --verbose: member nào giữ partition nào; phát hiện member không có việc.
• --state: STATE của group (Stable, đang rebalance…) và #MEMBERS.
• --reset-offsets: chỉ chạy khi consumer đã dừng; mặc định chỉ in ra, cần --execute mới thực hiện.

CLI hợp cho điều tra tay. Dashboard và alert cần exporter đọc cùng dữ liệu (committed offset + end offset) qua Admin API theo chu kỳ.

Metric phía consumer client

MBean kafka.consumer:type=consumer-fetch-manager-metrics (Java client, theo docs Monitoring 4.2):

Metric	Phạm vi	Ý nghĩa
records-lag-max	client	Lag lớn nhất (theo số record) của bất kỳ partition nào trong cửa sổ
records-lag, records-lag-avg, records-lag-max	từng partition (tag topic, partition)	Lag mới nhất / trung bình / lớn nhất của partition
records-lead-min, records-lead	client / partition	Khoảng cách từ vị trí đọc tới đầu log; gần 0 là sắp bị retention xoá
records-consumed-rate	client	Record/giây consumer nhận

MBean consumer-coordinator-metrics và consumer-metrics cho phần “vì sao”:

Metric	Dùng để
rebalance-rate-per-hour, rebalance-total	Phát hiện rebalance storm
time-between-poll-max, last-poll-seconds-ago	Handler chậm, sắp vượt max.poll.interval.ms
poll-idle-ratio-avg	Gần 0: consumer dành gần hết thời gian cho code xử lý, không phải chờ dữ liệu
commit-latency-avg, commit-latency-max	Commit chậm

Ba điểm khác biệt so với lag từ CLI:

1. Docs ghi rõ records-lag-max tính theo current position, không theo committed offset. Position là chỗ client đã fetch tới, có thể đi trước phần đã xử lý xong.
2. Với isolation.level=read_committed, client tính lag tới LSO; với read_uncommitted, tới high watermark (theo SubscriptionState.partitionLag).
3. Metric chỉ cập nhật khi client fetch. Consumer chết hoặc kẹt trong handler thì metric đứng yên hoặc biến mất, đúng lúc bạn cần nó nhất.

Kết luận: alert dựa trên lag tính từ committed offset bên ngoài consumer; metric client dùng để chẩn đoán.

Vì sao lag theo số record đánh lừa

Tình huống	Lag theo offset nói	Thực tế
Producer ghi burst	Rất lớn	Xả xong trong vài giây
Partition ít traffic, consumer kẹt	Rất nhỏ	Record đã chờ hàng phút
Topic compacted	Lớn hơn số record thật	Offset có khoảng trống sau compaction (compaction)
Topic có transaction	Không về 0 tuyệt đối	Control marker chiếm offset; transaction mở giữ LSO lại
Consumer read_committed, transaction mở lâu	CLI (đo tới HW) thấy lag tăng	Consumer đang chờ LSO, không phải chậm
Partition throughput khác nhau	Ngưỡng chung vô nghĩa	1.000 offset là 1 giây ở partition này, 1 giờ ở partition kia

Mô phỏng trong RAM (không có Kafka), mỗi giây producer ghi rồi consumer xử lý theo năng lực cố định:

npx --prefix examples ts-node examples/src/kafka-lag-sim/lag-sim.ts

Trích output thật:

t(s) | partition           | log-end | committed | lag(offset) | lag(giây)
  10 | P1 burst            |    6500 |      1100 |        5400 |         0
  15 | P1 burst            |    6750 |      4100 |        2650 |         5
  20 | P1 burst            |    7000 |      7000 |           0 |         0
  60 | P2 ít traffic, kẹt  |       7 |         1 |           6 |        50

P1 nhận burst 6.000 record ở giây 10, xả xong trước giây 20. P2 có handler kẹt từ giây 5: lag chỉ 6 offset nhưng record cũ nhất đã chờ 50 giây. Ngưỡng “lag > 1.000” kêu sai ở P1 và im lặng ở P2.

Time lag: SLI nên dùng

Time lag của partition = now − timestamp của record tại committed offset (record cũ nhất chưa xử lý). Bằng 0 khi lag offset bằng 0.

Nó trả lời đúng câu người dùng quan tâm: “event trễ bao lâu?”. Nó so sánh được giữa partition nhanh và chậm, và khớp trực tiếp với SLO (SLI/SLO).

Cách lấy timestamp:

• Exporter đọc committed offset, rồi fetch đúng record đó để lấy timestamp. Chính xác nhưng tốn một lần đọc mỗi partition mỗi chu kỳ.
• Hoặc nội suy: lưu cặp (log-end offset, thời điểm) theo chu kỳ, rồi ước lượng committed offset tương ứng với thời điểm nào. Rẻ hơn, kém chính xác khi traffic thất thường.

Lưu ý về timestamp:

• message.timestamp.type mặc định CreateTime: timestamp do producer đặt, phụ thuộc đồng hồ producer và có thể không tăng đều trong partition.
• Broker từ chối record có timestamp vượt đồng hồ broker quá message.timestamp.after.max.ms (mặc định 1 giờ); chiều quá khứ (before.max.ms) mặc định không giới hạn.
• Cần đồng hồ chính xác hơn thì dùng LogAppendTime ở topic, đổi lại mất thời điểm tạo event gốc.

Đo thêm end-to-end latency trong consumer (now − record.timestamp() khi xử lý xong) làm histogram. Nó bổ sung chứ không thay time lag: consumer kẹt thì không xử lý record nào, histogram im lặng.

Alert: ba quy tắc

Quy tắc	Điều kiện	Bắt được
Time lag vượt SLO, kéo dài	Time lag của partition xấu nhất vượt ngưỡng liên tục N phút	Consumer chậm hoặc kẹt, kể cả partition ít traffic
Lag tăng liên tục	Đạo hàm lag (offset/giây) dương trong một cửa sổ dài, ví dụ 10–15 phút	Năng lực consumer thấp hơn tốc độ ghi, trước khi vượt SLO
Không có tiến độ	Committed offset không đổi trong khi log-end offset tăng	Consumer chết, group trống, poison message chặn partition

Cách đặt cho đỡ nhiễu:

• Lấy max theo partition, không lấy tổng hay trung bình. Group by group, topic; giữ label partition chỉ khi số partition nhỏ (cardinality).
• Luôn có điều kiện kéo dài (for: vài phút). Burst ngắn tự hết.
• Đặt ngưỡng time lag từ SLO nghiệp vụ, không từ con số “có vẻ to”.
• Thêm cảnh báo riêng cho lead nhỏ: time lag tiến gần retention.ms nghĩa là sắp mất dữ liệu chưa đọc.

Phác thảo (tên metric tuỳ exporter bạn dùng):

# Pseudo-PromQL: thay tên metric theo exporter
- alert: KafkaConsumerTimeLagHigh
  expr: max by (group, topic) (consumer_group_time_lag_seconds) > 60
  for: 5m
- alert: KafkaConsumerNoProgress
  expr: |
    max by (group, topic, partition) (delta(consumer_group_committed_offset[10m])) == 0
    and max by (group, topic, partition) (delta(topic_partition_end_offset[10m])) > 0
  for: 5m

Chẩn đoán: lag tăng thì nhìn gì

Đang tải diagram…

Nguyên nhân thường gặp và cách sửa

Nguyên nhân	Dấu hiệu	Cách sửa
Handler chậm	poll-idle-ratio-avg thấp, mọi partition chậm đều	Profile handler; batch ghi DB; bỏ việc không cần đồng bộ
Rebalance storm	rebalance-rate-per-hour cao, CONSUMER-ID thường trống	Cooperative / group.protocol=consumer, static membership (rebalance)
Vượt max.poll.interval.ms (mặc định 5 phút)	time-between-poll-max chạm ngưỡng, member bị loại liên tục	Giảm max.poll.records (mặc định 500), tăng interval theo batch tệ nhất
Hot partition / key skew	Một partition lag, các partition khác 0	Đổi key hoặc tách key nóng (message key)
Consumer nhiều hơn partition	--members --verbose có member không có partition	Tăng partition (cẩn thận mapping key) hoặc bớt consumer
Downstream DB/API bão hoà	Latency DB tăng cùng lúc, lỗi timeout	Giới hạn concurrency, sửa query/index; đừng scale consumer
Poison message	Committed offset đứng yên trên một partition	Giới hạn retry, đưa sang dead-letter topic, log offset bị bỏ
Transaction mở lâu (read_committed)	Lag CLI tăng, consumer không bận	Tìm producer giữ transaction, kiểm tra transaction.timeout.ms

Khi nào thêm consumer không giúp gì

Thêm consumer chỉ tăng throughput khi cả hai điều kiện đúng: còn partition chưa có consumer riêng, và bottleneck nằm ở CPU/IO của chính consumer. Ngoài ra:

• Số consumer đã bằng số partition: consumer mới không có partition nào.
• Hot partition: một partition chỉ có một consumer trong group; lag của nó không đổi.
• Downstream bão hoà: thêm consumer là thêm connection và query vào DB đang quá tải, lag tăng nhanh hơn.
• Rebalance storm: mỗi lần thêm member là một lần rebalance nữa.
• Poison message hoặc transaction mở: partition bị chặn bởi một record hay LSO, không phải thiếu người xử lý.

Cần nhiều worker hơn số partition cho job độc lập: xem share groups.

Ví dụ thực tế

Alert theo tổng lag kêu oan, bỏ sót sự cố thật

Topic invoice-events 12 partition, group billing 12 consumer. Alert cũ: tổng lag lớn hơn 10.000 trong 1 phút.

• Mỗi sáng job đối soát ghi dồn khoảng 80.000 event trong vài giây; alert kêu, tự hết trong một phút.
• Một hôm partition 7 gặp event có payload lỗi, handler retry vô hạn. Partition này chỉ nhận vài chục event mỗi phút nên tổng lag vẫn dưới 10.000. Hoá đơn của nhóm khách trên partition 7 trễ ba giờ.

Đội đổi sang hai alert: time lag của partition xấu nhất lớn hơn 2 phút kéo dài 5 phút, và committed offset không tiến trong 10 phút. Handler giới hạn 5 lần retry rồi đẩy sang dead-letter topic. (Số liệu là kịch bản minh hoạ.)

Trade-offs

Alert theo time lag

• Khớp trực tiếp SLO nghiệp vụ (“event trễ bao lâu”).
• So sánh được giữa partition nhanh và partition ít traffic.
• Không kêu oan khi producer ghi burst.

Giá phải trả

• Cần exporter đọc timestamp hoặc nội suy; không có sẵn trong metric client.
• Phụ thuộc timestamp của producer khi dùng CreateTime.
• Label theo partition làm tăng số series cần lưu.

Checklist production

1. Đo lag bên ngoài consumer từ committed offset (exporter qua Admin API), không chỉ dựa vào records-lag-max của client.
2. Alert theo time lag của partition xấu nhất, có điều kiện kéo dài; ngưỡng lấy từ SLO.
3. Thêm alert không có tiến độ: committed offset đứng yên trong khi log-end offset tăng.
4. Dashboard có: lag theo partition, rebalance-rate-per-hour, time-between-poll-max, poll-idle-ratio-avg, latency downstream.
5. Theo dõi time lag so với retention.ms để không mất dữ liệu chưa đọc.
6. Giới hạn retry cho mỗi record và có dead-letter topic, để một record lỗi không chặn cả partition.
7. Trước khi scale consumer: kiểm tra số partition, key skew và tải của downstream.

Sai lầm thường gặp

Sai lầm thường gặp

Alert theo tổng lag của group. Tổng che partition kẹt và phóng đại burst. Dùng max theo partition, đo bằng thời gian.

Sai lầm thường gặp

Chỉ dựa vào metric của consumer client. Consumer chết thì metric biến mất; dashboard trông “sạch” đúng lúc sự cố. Luôn có nguồn lag từ committed offset phía broker.

Sai lầm thường gặp

Autoscale consumer theo lag. Lag tăng do downstream chậm hay hot partition thì scale vô ích hoặc có hại, và mỗi lần scale kích hoạt rebalance. Giới hạn số replica tối đa bằng số partition.

Tóm tắt

Tóm tắt

• Lag = end offset − committed offset, tính từng partition; đơn vị là offset, không hẳn là record.
• kafka-consumer-groups.sh --describe đo tới high watermark; records-lag của client đo từ position, tới LSO nếu read_committed.
• Lag theo số record kêu oan khi burst và im lặng khi partition ít traffic bị kẹt.
• SLI nên dùng là time lag của partition xấu nhất, alert có điều kiện kéo dài và kèm alert “không có tiến độ”.
• Thêm consumer chỉ giúp khi còn partition trống và bottleneck nằm ở chính consumer.

Tự kiểm tra: (1) Partition có lag 5 offset nhưng record cũ nhất 20 phút tuổi nói lên điều gì? (2) Vì sao records-lag-max có thể bằng 0 trong khi CLI vẫn thấy lag? (3) DB downstream đang 100% CPU, thêm consumer có giảm lag không?

Tham khảo

• Basic Kafka Operations — Apache Kafka 4.2
• Monitoring — Apache Kafka 4.2
• Consumer Configs — Apache Kafka 4.2
• Topic Configs — Apache Kafka 4.2
• ConsumerGroupCommand.java — Apache Kafka (GitHub)