Kafka transaction và "exactly once": output topic và offset cùng commit Nâng cao

Vấn đề

Processor đọc OrderPaid ở offset 7, ghi InvoiceCreated sang topic khác. Publish xong nhưng chết trước khi commit offset input: consumer mới đọc lại offset 7 và publish hóa đơn lần hai. Đảo thứ tự thì crash ở giữa làm sót hóa đơn (bài commit offset).

Khi cả input offset lẫn output đều nằm trong Kafka, transaction gộp chúng vào một ranh giới commit. Phạm vi của “exactly once” là: đọc Kafka, xử lý, ghi Kafka (cộng offset input). Ghi PostgreSQL, gọi payment API hay gửi email nằm ngoài.

Cơ chế

Một transaction Kafka bao output và tiến độ input

Phóng to sơ đồ

Input → Processoroffset 7

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

Processor đọc record offset 7; committed position của group vẫn là 7.

• transactional.id ổn định giúp Kafka nhận diện phiên và fence (chặn) producer cũ khi instance mới thay thế.
• Trong transaction: produce output, rồi gửi offset input (vị trí đọc tiếp, tức 8 sau record 7) vào transaction, rồi commit.
• Commit: output và offset cùng được xác nhận. Abort: output coi như aborted, offset không tiến, input được đọc lại.
• Idempotent producer là nền; transaction thêm tính nguyên tử cho nhóm write và offset.
• Consumer đọc input phải đặt enable.auto.commit=false; sau abort, ứng dụng phải reset vị trí đọc (abort không tự đưa con trỏ in-memory về offset cũ).

Tình huống	read_committed thấy output?	Offset input tiến?
Abort	Không	Không
Commit	Có	Có, cùng transaction
Ghi PostgreSQL rồi Kafka abort	Không	DB đã ghi: lệch hệ thống

LSO: transaction mở chặn cả partition

Consumer read_committed chỉ đọc tới LSO (last stable offset: offset đầu tiên của transaction còn mở, hoặc cuối log nếu không có). Mọi record sau LSO bị giữ lại, kể cả record thường đã commit từ producer khác.

Transaction mở làm consumer read_committed chờ

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

Producer A mở transaction và ghi record offset 10. Transaction chưa kết thúc.

Hệ quả vận hành: transaction mở lâu làm tăng độ trễ đọc cho cả partition. Broker tự abort transaction quá transaction.timeout.ms, nên đặt nó đủ ngắn.

Cheat-sheet config

Config	Mặc định	Ghi chú
transactional.id (producer)	null	Không đặt thì không dùng được transaction
transaction.timeout.ms (producer)	60000	Vượt transaction.max.timeout.ms của broker sẽ bị từ chối
transaction.max.timeout.ms (broker)	900000 (15 phút)	Trần cho timeout
isolation.level (consumer)	read_uncommitted	Phải đặt read_committed ở consumer downstream
enable.auto.commit (consumer input)	true	Đặt false trong processor transactional
transaction.state.log.replication.factor (broker)	3	Cluster dev một broker phải hạ xuống 1
transaction.state.log.min.isr (broker)	2	Như trên; không copy cấu hình dev lên production

Từ Kafka 4.0, giao thức transaction mạnh hơn (KIP-890) được bật khi finalize upgrade: producer epoch tăng theo từng transaction. Docs upgrade khuyên ứng dụng coi cả TimeoutException và TransactionAbortableException là lý do để abort transaction.

Khuyến nghị thực tế

• Dùng Kafka Streams với processing.guarantee=exactly_once_v2 thay vì tự viết vòng consume-transform-produce: thư viện lo fencing, rebalance, reset vị trí.
• Tự viết bằng Java consumer/producer chỉ khi bạn chấp nhận quản lý group metadata, epoch và lỗi commit/abort.
• Khi processor còn ghi PostgreSQL: dùng idempotent write theo eventId, hoặc outbox và inbox; Kafka transaction không khoá được DB.

Khi nào dùng / không dùng

Dùng	Không dùng
Pipeline Kafka đến Kafka (enrich, join, aggregate)	Đích cuối là DB hoặc API ngoài
Cần output nguyên tử qua nhiều partition/topic	Chỉ cần chống trùng retry (idempotent producer đủ)
Dùng Kafka Streams	Đội chưa vận hành nổi at-least-once cộng idempotent consumer

Ví dụ thực tế

Hai mức bảo đảm trong một pipeline

Processor chuyển OrderPaid thành InvoiceRequested, publish bằng producer transactional và commit offset input cùng transaction; consumer output dùng read_committed. Restart giữa batch thì downstream không thấy output aborted. Một service khác ghi invoice vào PostgreSQL với unique eventId và commit offset sau DB transaction. Đội ghi rõ: exactly-once ở luồng Kafka đến Kafka, at-least-once cộng idempotency ở Kafka đến PostgreSQL.

Mô hình trạng thái trong RAM (một mảng và một số; không tạo transaction marker hay fencing thật):

npx --prefix examples ts-node examples/src/kafka-transactions/transaction-model.ts

Output: attempt abort thì read_uncommitted thấy event, read_committed không thấy, offset input vẫn 7; sau lần thành công, read_committed thấy một invoice-created và offset input là 8. Repo chỉ có broker Kafka 3.6 một node, nên chưa chạy transaction thật.

Trade-offs

Ranh giới Kafka đến Kafka

• Output và offset input commit hoặc abort cùng nhau.
• read_committed che output của transaction aborted.
• Fencing chặn producer cũ ghi sau khi instance mới thay thế.

Giới hạn vận hành

• Phức tạp hơn at-least-once: abort, reset vị trí, rebalance, thời gian transaction.
• Transaction mở lâu tăng độ trễ đọc qua LSO.
• PostgreSQL, email, HTTP nằm ngoài transaction.

Checklist production

1. Chỉ dùng cho luồng Kafka đến Kafka; mọi chỗ khác dùng idempotency.
2. transactional.id ổn định và duy nhất theo instance/partition; enable.auto.commit=false cho input.
3. Consumer downstream đặt isolation.level=read_committed.
4. Giữ transaction.timeout.ms đủ ngắn; theo dõi độ trễ do LSO.
5. Production: transaction.state.log RF=3, min ISR=2.
6. Ưu tiên Kafka Streams exactly_once_v2.
7. Test crash giữa produce và commit, restart và rebalance.

Sai lầm thường gặp

Sai lầm thường gặp

Bật transaction nhưng vẫn auto-commit offset input riêng. Offset có thể đi trước output. Đưa offset vào transaction.

Sai lầm thường gặp

Downstream đọc read_uncommitted. Nó thấy cả record của transaction aborted. Đặt read_committed.

Sai lầm thường gặp

Gọi pipeline có DB là exactly-once end-to-end. Kafka không commit hộ DB. Định nghĩa ranh giới và thêm idempotency hoặc outbox.

Tóm tắt

Tóm tắt

• Transaction gộp output Kafka và offset input vào một commit.
• read_committed chỉ đọc tới LSO; transaction mở lâu làm chậm cả partition.
• Exactly-once không mở rộng sang PostgreSQL hay API ngoài.
• Đường thực tế nhất: Kafka Streams exactly_once_v2.

Tự kiểm tra: (1) Output đã produce nhưng transaction abort thì read_committed thấy gì? (2) Vì sao phải tắt auto-commit offset input? (3) Một transaction đang mở (chưa commit) ảnh hưởng consumer read_committed của partition đó thế nào?

Tham khảo

• Message Delivery Semantics and Using Transactions — Apache Kafka 4.2
• Producer Configs — Apache Kafka 4.2
• Upgrade notes — Apache Kafka 4.2