Kafka 3 — Consumer & Consumer Group

Ở bài trước ta đã hiểu producer đẩy record vào các partition của một topic. Bài này đi về phía đối diện: làm sao để nhiều tiến trình đọc dữ liệu đó vừa song song vừa không bỏ sót, không đọc trùng — và làm sao xử lý các tình huống hỏng hóc. Bối cảnh xuyên suốt: một hệ thống ngân hàng có topic transactions chứa các giao dịch, được nhiều hệ tiêu thụ độc lập cùng đọc (đối soát, chống gian lận, báo cáo, cập nhật số dư).

Mô hình Consumer Group

Consumer trong Kafka gần như luôn hoạt động trong một consumer group, xác định bởi cấu hình group.id. Group là đơn vị của cân bằng tải và theo dõi tiến độ đọc.

Quy tắc cốt lõi:

• Mỗi partition chỉ được đọc bởi đúng một consumer trong cùng một group tại một thời điểm. Điều này đảm bảo thứ tự đọc theo partition không bị phá vỡ bởi nhiều consumer cùng nhảy vào.
• Một consumer có thể đọc nhiều partition. Nếu group có ít consumer hơn số partition, một số consumer sẽ ôm nhiều partition.
• Số consumer hữu ích tối đa trong một group = số partition của topic. Nếu bạn có 6 partition và khởi động 8 consumer trong cùng group, 2 consumer sẽ rảnh rỗi (idle), không nhận partition nào. Đây là lý do khi thiết kế topic phải chọn số partition đủ lớn cho mức song song mong muốn — vì bạn không thể scale consumer vượt quá số partition.
• Nhiều group độc lập cùng đọc một topic. Group fraud-detection và group settlement đều đọc trọn vẹn transactions, mỗi group giữ offset riêng. Đây chính là mô hình publish/subscribe: một luồng dữ liệu, nhiều bên tiêu thụ không ảnh hưởng lẫn nhau.

Sơ đồ phân phối partition

Với 4 partition và 2 consumer, mỗi consumer nhận 2 partition. Nếu thêm Consumer C, một trong hai consumer hiện tại sẽ nhả bớt partition để C nhận — đó là quá trình rebalance mô tả bên dưới. Nếu thêm tới Consumer E (consumer thứ 5), nó sẽ đứng chờ vì không còn partition nào để giao.

Vì sao scale bằng partition

Trong hệ ngân hàng, nếu ta partition transactions theo account_id (dùng key = account_id), thì mọi giao dịch của cùng một tài khoản luôn rơi vào cùng partition, do đó luôn được xử lý tuần tự bởi cùng một consumer. Đây là điều kiện tiên quyết để cập nhật số dư đúng thứ tự. Muốn tăng thông lượng xử lý, ta tăng số partition và thêm consumer tương ứng — nhưng cần lưu ý việc tăng partition sẽ thay đổi ánh xạ key→partition với dữ liệu mới, và không di dời dữ liệu cũ.

Phân phối partition — Partition Assignor

Khi các thành viên trong group thay đổi, cần một thuật toán quyết định consumer nào giữ partition nào. Kafka cho cấu hình qua partition.assignment.strategy:

• RangeAssignor (mặc định lịch sử): với mỗi topic, chia dãy partition liên tiếp cho các consumer theo thứ tự. Nhược điểm: khi consumer đăng ký nhiều topic có ít partition, các consumer đầu tiên dễ bị dồn tải.
• RoundRobinAssignor: rải toàn bộ partition (của mọi topic đăng ký) lần lượt qua các consumer. Cân bằng tốt hơn nhưng mỗi lần rebalance có thể xáo trộn hầu hết assignment.
• StickyAssignor: cân bằng như round-robin nhưng cố gắng giữ nguyên assignment cũ qua các lần rebalance, giảm số partition bị di chuyển.
• CooperativeStickyAssignor: bản sticky hỗ trợ rebalance hợp tác (incremental cooperative). Đây là lựa chọn khuyến nghị cho hệ thống hiện đại vì tránh được "stop-the-world" (giải thích ở phần rebalance).

# Khuyến nghị cho ứng dụng mới
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Offset & Commit

Offset là số thứ tự tăng dần, duy nhất trong một partition, đánh dấu vị trí của mỗi record. Consumer không xóa dữ liệu khi đọc; nó chỉ ghi nhớ "đã đọc tới offset nào" để lần sau tiếp tục. Vị trí này gọi là committed offset.

__consumer_offsets

Kafka lưu committed offset trong một topic nội bộ tên __consumer_offsets (compacted). Mỗi bản ghi có key gồm (group.id, topic, partition) và value là offset đã cam kết. Nhờ vậy khi consumer khởi động lại hoặc partition được giao cho consumer khác, nó biết đọc tiếp từ đâu. Committed offset thường được hiểu là "offset của record kế tiếp cần đọc" (tức lastProcessed + 1).

Auto-commit vs Manual commit

Auto-commit (enable.auto.commit=true): client tự động commit định kỳ mỗi auto.commit.interval.ms (mặc định 5000ms), commit vị trí đã được poll() trả về. Đơn giản nhưng nguy hiểm: nó commit dựa trên đã fetch, không phải đã xử lý xong. Nếu bạn poll một lô, commit chạy ngầm, rồi ứng dụng crash giữa lúc xử lý — các record chưa xử lý xong đã bị coi là đã commit và sẽ mất (nghiêng về at-most-once). Với dữ liệu giao dịch ngân hàng, gần như luôn nên tắt auto-commit.

Manual commit (enable.auto.commit=false): bạn tự gọi commit sau khi xử lý xong. Hai kiểu:

• commitSync(): chặn tới khi broker xác nhận, tự retry khi lỗi tạm thời. An toàn nhất, độ trễ cao hơn.
• commitAsync(): gửi rồi tiếp tục ngay, không tự retry (retry có thể ghi đè offset mới hơn bằng offset cũ). Thông lượng cao hơn nhưng cần xử lý callback lỗi.

Mẫu phổ biến: dùng commitAsync() trong vòng lặp bình thường để nhanh, và commitSync() một lần cuối khi đóng consumer để đảm bảo chốt sạch.

group.id=settlement-service
enable.auto.commit=false
auto.offset.reset=earliest        # earliest | latest | none — khi chưa có committed offset
max.poll.records=500              # số record tối đa mỗi lần poll()
max.poll.interval.ms=300000       # thời gian tối đa giữa 2 lần poll trước khi bị coi là chết
session.timeout.ms=45000          # thời gian không heartbeat trước khi bị loại khỏi group
heartbeat.interval.ms=3000        # nhịp gửi heartbeat

Lưu ý auto.offset.reset chỉ áp dụng khi không có committed offset (group mới, hoặc offset cũ đã bị hết hạn/xóa): earliest đọc từ đầu, latest đọc từ record mới tới, none ném lỗi.

Vòng lặp Poll và các timeout

Consumer hoạt động theo poll loop: liên tục gọi poll(timeout). Một poll() làm hai việc: (1) lấy record về, (2) đồng thời điều phối heartbeat và phối hợp với group coordinator.

Hai timeout dễ nhầm lẫn:

• session.timeout.ms: nếu coordinator không nhận heartbeat trong khoảng này, coi consumer đã chết và kích hoạt rebalance. Heartbeat được gửi ngầm trên một thread riêng theo heartbeat.interval.ms.
• max.poll.interval.ms: khoảng thời gian tối đa giữa hai lần gọi poll(). Nếu bạn xử lý một lô quá lâu (ví dụ gọi service bên ngoài chậm) mà chưa poll lại, consumer bị coi là "kẹt" và bị loại khỏi group — dù heartbeat vẫn đều. Đây là nguyên nhân phổ biến của rebalance bất ngờ. Cách xử lý: giảm max.poll.records, tăng max.poll.interval.ms, hoặc đẩy việc xử lý nặng sang thread khác một cách cẩn thận.

Rebalance

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

• Một consumer tham gia (khởi động thêm instance) hoặc rời group (tắt, crash, hoặc quá session.timeout.ms/max.poll.interval.ms).
• Số partition của topic thay đổi (tăng partition).
• Danh sách topic đăng ký (khi dùng pattern subscribe) thay đổi.

Eager (stop-the-world) vs Cooperative (incremental)

• Eager rebalance (giao thức cũ, ví dụ Range/RoundRobin/Sticky classic): tất cả consumer nhả toàn bộ partition đang giữ, rồi mới nhận lại assignment mới. Trong khoảng đó cả group ngừng xử lý — gọi là "stop-the-world". Với nhiều partition và group lớn, khoảng dừng này gây tăng lag đột biến.
• Cooperative / incremental rebalance (CooperativeStickyAssignor): chỉ những partition cần đổi chủ mới bị nhả, phần còn lại tiếp tục được xử lý bình thường. Rebalance diễn ra qua nhiều nhịp nhỏ, giảm gián đoạn đáng kể. Đây là lý do nên chuyển sang cooperative-sticky cho các dịch vụ nhạy với độ trễ như đối soát giao dịch.

Static membership

Trong triển khai container/Kubernetes, mỗi lần restart pod, consumer thường được cấp identity mới và gây rebalance. Static membership khắc phục bằng cách gán group.instance.id cố định cho từng instance:

group.instance.id=settlement-worker-1
session.timeout.ms=45000

Khi instance có group.instance.id restart nhanh và quay lại trước khi hết session.timeout.ms, coordinator giữ nguyên assignment cho nó, không kích hoạt rebalance. Rất hữu ích khi rolling update thường xuyên.

ConsumerRebalanceListener

Khi rebalance xảy ra, bạn nên commit offset trước khi mất partition để tránh xử lý lại quá nhiều. Client cung cấp callback:

• onPartitionsRevoked: gọi trước khi nhả partition — nơi commit offset cuối.
• onPartitionsAssigned: gọi sau khi nhận partition mới — nơi khôi phục trạng thái nếu cần.

Delivery Semantics từ góc consumer

Ngữ nghĩa giao nhận phụ thuộc vào thứ tự giữa "xử lý" và "commit offset".

At-least-once (mặc định, khuyến nghị)

Xử lý record trước, commit offset sau. Nếu crash giữa hai bước, sau khi khởi động lại consumer đọc lại từ committed offset cũ và xử lý lại một số record. Không mất dữ liệu, nhưng có thể trùng. Đây là mặc định thực dụng cho hầu hết hệ thống — miễn là phía xử lý idempotent.

At-most-once

Commit offset trước, xử lý sau (hoặc dùng auto-commit vô tư). Nếu crash sau commit trước khi xử lý xong, record đó mất luôn. Chỉ chấp nhận được cho dữ liệu không quan trọng (ví dụ metric có thể mất vài điểm). Không phù hợp với giao dịch tài chính.

Exactly-once (EOS) — read-process-write

Khi ứng dụng đọc từ Kafka, xử lý, rồi ghi kết quả trở lại Kafka, có thể đạt exactly-once bằng transaction: gói cả việc ghi record kết quả và việc commit offset nguồn vào cùng một giao dịch. Nếu giao dịch abort, cả kết quả lẫn offset đều không được ghi nhận.

# Phía consumer đọc kết quả EOS
isolation.level=read_committed    # chỉ đọc record đã commit của transaction

# Phía producer trong luồng read-process-write
enable.idempotence=true
transactional.id=settlement-processor-1

Điểm mấu chốt: EOS của Kafka là exactly-once trong phạm vi Kafka-to-Kafka. Nếu "process" bao gồm ghi ra hệ ngoài (DB số dư), thì exactly-once thật sự cần thêm cơ chế idempotent ở phía DB — Kafka transaction không tự động phủ ra hệ ngoài.

Idempotent consumer & xử lý trùng

Vì at-least-once có thể sinh bản trùng, phía xử lý nên idempotent: xử lý cùng một record nhiều lần cho cùng kết quả. Kỹ thuật thường dùng với giao dịch ngân hàng:

• Dedup theo khóa nghiệp vụ: mỗi giao dịch mang transaction_id duy nhất; trước khi ghi số dư, kiểm tra transaction_id đã xử lý chưa (bảng processed_ids hoặc ràng buộc UNIQUE).
• Upsert theo khóa thay vì insert mù, để lần chạy lại không nhân đôi.
• Ghi kèm offset đã xử lý vào chính DB đích trong một transaction DB, để nguồn sự thật về tiến độ nằm ở DB (kỹ thuật này còn cho phép khôi phục offset từ DB thay vì từ Kafka).

Ví dụ code Consumer

Python (confluent-kafka), at-least-once + idempotent

from confluent_kafka import Consumer, KafkaException

conf = {
    "bootstrap.servers": "broker1:9092,broker2:9092",
    "group.id": "settlement-service",
    "enable.auto.commit": False,
    "auto.offset.reset": "earliest",
    "partition.assignment.strategy": "cooperative-sticky",
    "max.poll.interval.ms": 300000,
    "session.timeout.ms": 45000,
}

consumer = Consumer(conf)
consumer.subscribe(["transactions"])

seen = set()  # thực tế nên là bảng bền vững / ràng buộc UNIQUE trong DB

try:
    while True:
        msg = consumer.poll(1.0)          # timeout 1s
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())

        txn_id = msg.key().decode() if msg.key() else None
        # Idempotent: bỏ qua nếu đã xử lý (chống trùng của at-least-once)
        if txn_id not in seen:
            process_transaction(msg.value())   # xử lý TRƯỚC
            seen.add(txn_id)

        # Commit SAU khi xử lý xong -> at-least-once
        consumer.commit(message=msg, asynchronous=False)   # commitSync
finally:
    consumer.commit(asynchronous=False)   # chốt sạch trước khi đóng
    consumer.close()

Java, xử lý theo lô + commit sau lô

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "settlement-service");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
        "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("transactions"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> r : records) {
            processTransaction(r.value());   // xử lý trước
        }
        if (!records.isEmpty()) {
            consumer.commitAsync();          // commit sau lô, nhanh
        }
    }
} catch (WakeupException ignored) {
    // shutdown bình thường
}

Consumer Lag

Consumer lag của một partition = log end offset (offset record mới nhất broker đã ghi) trừ committed offset của group. Nói cách khác: consumer đang chậm bao nhiêu record so với đầu partition. Lag của group thường lấy tổng (hoặc max) trên tất cả partition.

Vì sao quan trọng:

• Lag tăng đều đặn nghĩa là tốc độ tiêu thụ chậm hơn tốc độ sản xuất — hệ đang tụt lại, độ trễ dữ liệu end-to-end tăng. Với đối soát/chống gian lận ngân hàng, lag cao đồng nghĩa phát hiện bất thường bị trễ.
• Lag đột biến thường là dấu hiệu của rebalance stop-the-world, consumer chết, hoặc xử lý bị nghẽn.
• Lag giúp đánh giá có cần thêm consumer / thêm partition hay không.

Cách đo:

# Xem lag theo từng partition của một group
kafka-consumer-groups.sh --bootstrap-server broker1:9092 \
  --describe --group settlement-service
# Cột LAG = LOG-END-OFFSET - CURRENT-OFFSET cho mỗi partition

Trong vận hành thực tế, người ta dùng công cụ như Burrow hoặc Kafka Exporter (xuất metric ra Prometheus) và đặt cảnh báo. Nên cảnh báo dựa trên xu hướng lag (lag tăng liên tục nhiều phút) hơn là một ngưỡng tuyệt đối tức thời, vì lag nhất thời sau khi deploy là bình thường. Một tín hiệu bổ trợ tốt là time lag — record cũ nhất chưa xử lý bị chậm bao nhiêu giây, phản ánh độ trễ nghiệp vụ trực tiếp hơn số record.

Tóm tắt

• Consumer làm việc theo group; trong một group, mỗi partition chỉ do một consumer đọc, nên số partition là trần scale của group. Nhiều group đọc độc lập cùng một topic (pub/sub).
• Assignor quyết định ai giữ partition nào; CooperativeStickyAssignor là lựa chọn khuyến nghị vì hỗ trợ rebalance hợp tác, tránh stop-the-world.
• Offset ghi tiến độ, lưu trong __consumer_offsets. Nên tắt auto-commit và commit thủ công sau khi xử lý để đạt at-least-once an toàn.
• Rebalance xảy ra khi thành viên/partition thay đổi; chú ý session.timeout.ms (heartbeat) và max.poll.interval.ms (thời gian giữa 2 poll); dùng group.instance.id (static membership) để tránh rebalance khi restart.
• Delivery semantics: at-least-once là mặc định thực dụng (cần idempotent để chống trùng); exactly-once đạt được bằng transaction trong luồng read-process-write nội bộ Kafka.
• Consumer lag là chỉ số sức khỏe quan trọng nhất của phía tiêu thụ; theo dõi xu hướng và cảnh báo sớm.

Tự kiểm tra

1. Một topic có 4 partition, group có 6 consumer — bao nhiêu consumer thực sự nhận dữ liệu, và vì sao?
2. Tại sao auto-commit có thể dẫn tới mất record thay vì xử lý trùng? Thứ tự sự kiện nào gây ra điều đó?
3. Phân biệt vai trò của session.timeout.ms và max.poll.interval.ms. Nếu consumer xử lý mỗi lô mất 6 phút mà max.poll.interval.ms=300000, điều gì xảy ra?
4. Rebalance eager (stop-the-world) khác cooperative/incremental ở điểm nào, và tại sao cooperative giảm được lag đột biến?
5. Vì sao at-least-once cần consumer idempotent, và nêu một kỹ thuật chống trùng cho giao dịch ngân hàng.
6. Consumer lag được tính như thế nào, và vì sao nên cảnh báo theo xu hướng lag thay vì một ngưỡng tuyệt đối tức thời?

Đọc tiếp

Kafka 4 — Lưu trữ & Độ tin cậy