Kafka Internals #2: send() không gửi record — producer, retry và giới hạn của exactly-once

Kafka Internals — phần 2/3. Phần 1 định nghĩa log, partition, offset và ISR. Phần này đi theo một record từ send() đến broker. Phần 3 đi vào consumer, offset commit và rebalance.

Phiên bản: Apache Kafka 4.3.x, Java client 4.3.x, đối chiếu ngày 2026-10-25. Defaults thay đổi theo phiên bản; các giá trị dưới đây không áp nguyên cho Kafka 3.x. Những chỗ protocol hoặc default đã đổi sẽ được ghi rõ.

Giả định: topic orders, 12 partitions, replication factor 3, min.insync.replicas=2, key orderId, event OrderPlaced khoảng 1 KiB; ba broker; producer và consumer chạy trên JVM; lưu lượng đỉnh 50.000 record/giây; p99 produce latency mục tiêu 50 ms. Các con số là giả định để phân tích, không phải kết quả đo.

Future trả về, record vẫn nằm trong RAM

Một service đặt hàng gọi producer.send(record) và nhận về một Future trong vài micro giây. Process bị SIGKILL trước khi Sender kịp gửi. Broker chưa từng thấy record; lời gọi API đã trả về thành công.

Chiều ngược lại cũng tồn tại. Broker đã append record, ack bị mất trên đường về, producer retry. Nếu producer không idempotent, record được ghi hai lần và payment service bị gọi hai lần. Một bên là mất, một bên là trùng; cả hai xảy ra sau khi send() đã trả về.

Đây là hai ví dụ giả định, nhưng cơ chế thì có thật. Chúng chỉ ra rằng send() không phải một lời hứa giao nhận. Nó là một thao tác trên bộ nhớ. ACK của broker chỉ cho biết kết quả của lần produce theo acks và trạng thái ISR đã chọn; nó không đảm bảo record đã fsync, không đảm bảo consumer đã xử lý, và không đảm bảo side effect ngoài Kafka. Một guarantee đầy đủ có ba lớp: producer delivery, Kafka durability, và end-to-end processing. Câu hỏi của bài này: record đi qua những trạm nào, cửa sổ nào có thể mất hoặc trùng, và exactly-once đóng được bao nhiêu cửa sổ trong số đó?

Bài toán: không mất đơn hàng, không trùng thanh toán

Giả định lưu lượng đỉnh 50.000 record/giây, mỗi record khoảng 1 KiB. Tổng ingress xấp xỉ 50 MB/s trước nén; với replication factor 3, mỗi record được ghi ba lần nên tổng lưu lượng ghi trên mạng — produce cộng hai chặng replicate — vào khoảng 150 MB/s. Chia đều trên 12 partitions, mỗi partition nhận khoảng 4.200 record/giây. Đây là số để suy luận, không phải kết quả đo.

Ràng buộc nghiệp vụ rõ hơn con số: mất một event đơn hàng là lỗi; gọi payment hai lần cho cùng một đơn cũng là lỗi; p99 produce latency phải nằm trong 50 ms; downstream chấp nhận eventual consistency trong vài giây. Thứ tự theo orderId phải giữ, vì các event của cùng một đơn được xử lý như một chuỗi.

Ba cách sửa ngây thơ đều thất bại vì bỏ qua cơ chế. Gọi send().get() cho từng record biến mỗi record thành một round-trip đến broker. Với throughput 50.000 record/giây và latency trung bình giả định 50 ms, Little's law cho khoảng 2.500 thao tác đang chờ đồng thời — đây là ước lượng concurrency, không phải số request broker phải xử lý đồng thời, và p99 có thể cao hơn latency trung bình. Mọi dao động của broker trở thành latency của API. Đặt acks=1 để nhanh hơn sẽ mất record khi leader chết trước lúc follower replicate. Đặt retries=0 để "tránh trùng" biến mọi lỗi mạng tạm thời thành mất dữ liệu.

Mục tiêu của thiết kế, theo thứ tự: durability và ordering theo key trước, latency sau, throughput cuối. Nếu thứ tự mục tiêu khác, phần lớn khuyến nghị trong bài cũng đổi theo.

send() chỉ là enqueue; Sender mới là nơi record rời process

KafkaProducer là một object dùng chung, thread-safe. Nó giữ metadata của cluster, serializer, interceptor và một buffer. Tạo producer cho mỗi request là cách chắc chắn nhất để mất batching, tạo producer ID mới và không dedupe được với session trước.

Khi gọi send(record, callback), producer chạy interceptor, serialize key và value, rồi chọn partition. Với key, partition là hash của key; cùng key vào cùng partition. Với key null, partitioner dùng sticky partition: dồn record vào một partition cho đến khi đủ batch.size, sau đó đổi partition. Custom partitioner thay được cả hai hành vi, nhưng phải tự chịu trách nhiệm về phân bố.

Record sau đó được append vào RecordAccumulator. Accumulator giữ ProducerBatch theo từng partition; BufferPool cấp phát bộ nhớ theo batch.size. Nếu buffer cạn, send() block tới max.block.ms rồi trả về một future fail với BufferExhaustedException — không phải ném exception ngay. max.block.ms cũng là trần cho lần gửi đầu tiên khi chưa có metadata của topic.

Sender thread mới là nơi record rời process. Nó lấy các batch đã sẵn sàng, gom theo broker leader, tạo ProduceRequest, gửi qua NetworkClient và giữ tối đa max.in.flight.requests.per.connection request trên mỗi connection. Batch lỗi được đưa lại vào accumulator để retry nếu còn trong delivery.timeout.ms.

Mỗi partition có một deque batch riêng. Khi batch đầu của một partition đang retry, các batch sau của cùng partition không được vượt qua nếu producer còn giữ ordering; với idempotence, broker cũng từ chối sequence sai thứ tự. Partition khác không bị ảnh hưởng. Vì vậy một key nóng có thể có p99 cao hơn hẳn phần còn lại của cluster dù broker không quá tải — và đó là lý do partition key là quyết định khó đảo ngược.

send() trả về FutureRecordMetadata ngay sau khi record nằm trong buffer, trừ hai trường hợp có thể block: lần gửi đầu tiên cần metadata, và lúc cấp buffer. Future.get() chỉ hoàn tất khi broker ack theo acks, hoặc khi request thất bại; lỗi được bọc trong ExecutionException. RecordMetadata trả về partition, offset và timestamp do broker gán — offset đó chỉ có nghĩa trong partition, không phải số thứ tự toàn topic.

Callback chạy trên I/O thread của producer. Một callback chậm — gọi DB, gọi HTTP, ghi log đồng bộ — sẽ giữ Sender lại và làm chậm mọi partition, không chỉ partition của record đó. Code C1 đặt phần xử lý nặng ra khỏi callback và chỉ dùng flush() như một barrier trước khi tắt.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
    "broker1:9092,broker2:9092,broker3:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

KafkaProducer<String, String> producer =
    new KafkaProducer<>(props);

// send() chỉ đưa record vào buffer; kết quả nằm ở callback.
producer.send(
    new ProducerRecord<>("orders", orderId, payload),
    (metadata, error) -> {
        if (error != null) {
            // callback chạy trên I/O thread: không block
            log.error("produce failed for {}", orderId, error);
            return;
        }
        log.debug("orders-{} offset {}",
            metadata.partition(), metadata.offset());
    });

// flush() là barrier cho các record đã gửi trước đó.
producer.flush();
producer.close(Duration.ofSeconds(10));

Hai bề mặt lỗi của send() thường bị gộp làm một. Exception ném ngay trên thread gọi là lỗi cấu hình hoặc lập trình: serialize sai, chưa beginTransaction, producer đã close. Lỗi giao nhận thật — buffer cạn, metadata timeout, broker trả lỗi, delivery timeout — đến qua future hoặc callback. Tầng nào retry phụ thuộc vào việc tầng đó có biết record đã được ghi hay chưa.

flush() không phải "gửi ngay". Nó đưa các record đang buffer vào trạng thái sẵn sàng gửi và chờ chúng hoàn tất; record gửi sau lời gọi flush() không nằm trong guarantee. close() flush rồi giải phóng tài nguyên. Trong production, close(Duration) nên có trần thay vì chờ vô hạn khi broker không phản hồi.

Batch quyết định latency; throughput chỉ là hệ quả

Một batch rời accumulator khi đầy batch.size, khi hết linger.ms, hoặc khi Sender thấy có thể gửi. Tài liệu 4.3 ghi thêm một chi tiết dễ bỏ qua: backpressure từ broker có thể làm linger thực tế dài hơn giá trị cấu hình.

Với mặc định 4.3 — batch.size 16 KiB, linger.ms 5 ms, buffer.memory 32 MiB, compression.type=none, max.request.size 1 MiB — một record chịu hai khoảng thời gian: queue time trong accumulator và request time đến ack. Giả định request time đến broker trong cùng AZ khoảng 5–15 ms và ack mất thêm 5 ms; cộng 5 ms linger, p99 gần 25–30 ms. Ngân sách 50 ms còn lại 20 ms cho variance và retry. Đây là số giả định để thấy cấu trúc, không phải benchmark.

Đây chỉ là ngân sách sơ bộ: không thể cộng các giá trị danh nghĩa để suy ra p99 thực tế. p99 còn phụ thuộc phân bố latency, queueing, GC, số partition, broker saturation và retry; phải đo queue time, request latency và end-to-end latency dưới tải đại diện.

linger.ms=0 không xoá bỏ batching dưới tải. Record đến gần nhau vẫn được gom; điều thay đổi là producer không chờ thêm, nên số request tăng và queue time chuyển sang socket hoặc broker. Ở tải thấp, linger.ms=0 giảm latency; ở tải cao, nó có thể tăng latency vì nhiều request nhỏ cạnh tranh nhau.

Compression đánh đổi CPU lấy byte trên dây, và hiệu quả phụ thuộc kích thước batch: batch nhỏ làm tỷ lệ nén tệ đi. none là mặc định; lz4 thường là điểm cân bằng cho latency, zstd nén tốt hơn nhưng tốn CPU hơn. Không có lựa chọn đúng nếu chưa biết payload và CPU headroom.

Bộ nhớ cũng là một giới hạn vật lý. buffer.memory 32 MiB chia cho batch.size 16 KiB cho khoảng 2.000 batch đang giữ, nhưng batch thực tế có thể lớn hơn khi record lớn. Với 12 partitions và nhiều producer instance, buffer nhân theo số instance; đây là chỗ capacity planning hay bị bỏ qua.

Batching là per partition, nên phân bố key quyết định hiệu quả. Partition nóng lấp đầy batch nhanh; partition lạnh chờ hết linger.ms. Buffer dùng chung, nên partition nóng có thể chiếm phần lớn buffer.memory và làm send() cho partition khác block dù tổng tải chưa cao.

Chi phí cross-AZ cũng nằm trong ngân sách. acks=all chờ các replica trong ISR; nếu ba replica nằm ở ba AZ, mỗi record đi thêm hai chặng liên AZ. Với 50 MB/s ingress giả định, lưu lượng cross-AZ có thể lên hàng trăm MB/s và trở thành một trong những driver chi phí lớn nhất — trước cả CPU nén.

Các giá trị trong bảng là điểm khởi đầu để hiệu chỉnh, không phải cấu hình chuẩn.

Retry đóng cửa sổ mất và mở cửa sổ duplicate

Producer 4.3 mặc định retries=Integer.MAX_VALUE, delivery.timeout.ms=120.000, request.timeout.ms=30.000, retry.backoff.ms=100 và max.in.flight.requests.per.connection=5. delivery.timeout.ms là trần tổng từ lúc record vào buffer đến lúc báo thành công hoặc thất bại; nó phải lớn hơn hoặc bằng request.timeout.ms + linger.ms. Retry không vượt qua trần này.

acks quyết định ai đã có record trước khi producer coi là xong. acks=0 không chờ ai, retry không có tác dụng vì client không biết lỗi, và offset trả về là -1. acks=1 chỉ chờ leader ghi; leader chết trước khi follower replicate là mất. acks=all chờ toàn bộ ISR; nếu ISR nhỏ hơn min.insync.replicas, broker từ chối ghi thay vì nhận rủi ro.

Con số delivery.timeout.ms=120.000 và request.timeout.ms=30.000 tạo ra một ngân sách thử lại. Trong trường hợp xấu nhất, một record có thể đi qua khoảng bốn cửa sổ request timeout trước khi hết hạn, ít hơn nếu backoff hoặc xếp hàng chiếm chỗ. Đây là lý do tăng request.timeout.ms mà không tăng delivery.timeout.ms có thể làm giảm số lần thử chứ không tăng độ bền.

Không phải lỗi nào cũng đáng retry. NotLeaderForPartition, LeaderNotAvailable, NotEnoughReplicas, NetworkException là tạm thời và producer tự xử lý. RecordTooLargeException, lỗi serializer và lỗi cấu hình không tự khỏi; retry chỉ làm chậm lỗi. Phân biệt hai nhóm này trước khi viết logic retry ở tầng ứng dụng, vì tầng ứng dụng thường không có đủ thông tin để biết record đã vào log hay chưa.

Ba cửa sổ duplicate, reorder và loss cần tách riêng:

• Ack mất sau khi broker ghi. Producer không biết record đã vào log; retry gửi lại. Không idempotent thì duplicate; có idempotent thì broker dedupe trong cùng session.
• Leader đổi giữa request. Producer nhận NotLeaderForPartition hoặc LeaderNotAvailable, refresh metadata và gửi lại cho leader mới. Với acks=1, record có thể đã mất trước khi leader mới nhận.
• Retry với nhiều in-flight. Khi enable.idempotence=false và max.in.flight>1, batch sau có thể đến trước batch trước đang retry, làm đảo thứ tự. Đây là lý do tài liệu khuyến nghị idempotence hoặc giảm in-flight khi cần ordering.

max.in.flight là per connection, không phải per partition. Với 3 broker và 12 partitions, producer có thể có tới 15 request đang bay. Con số này ảnh hưởng đến bộ nhớ, cửa sổ duplicate và khả năng reorder, nên đừng suy luận từ con số 5 như một giới hạn toàn cục.

Một batch hết delivery.timeout.ms trả về TimeoutException. Ứng dụng retry bằng cách tạo record mới là tạo một duplicate mới, ngoài tầm producer: broker không biết hai record là cùng một ý định. Đây là điểm nối quan trọng sang phần idempotence và transaction.

Idempotence đóng cửa sổ duplicate — trong một session

Idempotent producer xin broker một producer ID (PID) và epoch, rồi gắn sequence number tăng dần cho từng batch trên từng partition. Broker giữ state theo cặp PID và partition, đủ để nhận ra batch gửi lại. Tài liệu 4.3 ghi rõ broker chỉ giữ tối đa 5 batch gần nhất cho mỗi producer; con số này giải thích vì sao max.in.flight bị chặn ở 5 khi bật idempotence.

Ba ràng buộc đi kèm: acks=all, retries>0, max.in.flight<=5. Nếu cấu hình xung đột mà idempotence không được bật rõ ràng, producer tự tắt idempotence; nếu bật rõ ràng, producer ném ConfigException thay vì chạy với guarantee khác. Từ Kafka 3.0, enable.idempotence=true là mặc định, nên phần lớn hệ thống 4.x đã có lớp bảo vệ này mà không cần làm gì.

Giới hạn nằm ở chữ session. Restart process tạo PID mới. Broker không còn state cũ, nên record gửi lại từ session trước không được dedupe. Idempotence cũng không bảo vệ side effect ngoài Kafka, không bảo vệ app-level retry tạo record mới, và không bảo vệ khi retry vượt quá cửa sổ 5 batch. Nó là một lớp chống duplicate hẹp, không phải EOS.

Broker state cũng có thể mất. Nếu leader bị truncate log hoặc một replica ngoài ISR được bầu làm leader, sequence của producer không còn khớp với state mà broker đang giữ. Cách xử lý khác nhau theo loại producer:

• Idempotent producer không transaction: Java client 4.x bump epoch cho partition và reset sequence để tiếp tục gửi. Producer không dừng, nhưng gap trong log là tín hiệu dữ liệu có thể đã mất; phải kiểm tra record nào đã vào log, không coi đó là retry trong suốt.
• Transactional producer: OutOfOrderSequenceException là fatal; phải đóng producer và điều tra.

Đây là correctness event, không phải lỗi mạng để retry mù.

Chi phí của idempotence thấp: thêm state trên broker theo PID và partition, giới hạn in-flight, và một chút latency. Cái giá này nhỏ so với việc ghi trùng, nên không có lý do tắt nó trong 4.x trừ khi có ràng buộc rất cụ thể về client hoặc broker cũ.

Transaction mở rộng guarantee qua session, rồi dừng ở biên Kafka

transactional.id định danh một producer logic qua nhiều lần restart. initTransactions() fence instance cũ, lấy PID và epoch mới, rồi đăng ký với transaction coordinator. Từ thời điểm đó, chỉ instance đang giữ epoch mới được ghi; instance cũ nhận ProducerFencedException nếu cố tiếp tục. Đây là cơ chế chống zombie, không phải một tính năng phụ.

Một producer chỉ có một transaction mở tại một thời điểm. Chuỗi API gồm beginTransaction(), nhiều lần send(), sendOffsetsToTransaction(offsets, consumerGroupMetadata), rồi commitTransaction() hoặc abortTransaction(). sendOffsetsToTransaction block đến khi consumer group coordinator nhận offset, nhưng offset chỉ được coi là commit khi transaction commit. Consumer phải tắt auto-commit và không tự commit offset trong lúc đó.

Coordinator ghi trạng thái vào topic nội bộ __transaction_state. Khi commit, nó gửi marker tới leader của từng partition tham gia; khi abort, marker abort được ghi và record bị ẩn với consumer read_committed. Consumer read_committed chỉ trả record tới last stable offset (LSO), là offset của transaction mở sớm nhất trừ một. Record aborted bị lọc, nhưng offset vẫn có khoảng trống.

Hệ quả availability ít được nói đến: một transaction mở lâu trên partition giữ LSO lại. Giả định transaction A mở ở offset 100 trên partition 3; một producer khác ghi offset 101–150; consumer read_committed vẫn dừng ở 99 cho tới khi A commit hoặc abort. Record của producer thứ hai không sai, nhưng không đọc được. Thời lượng transaction trở thành một tham số p99 của phía đọc.

Transaction chỉ bảo vệ consumer nếu consumer chọn đúng isolation. read_uncommitted là mặc định của consumer: nó trả cả record của transaction aborted. Một consumer quên đặt read_committed sẽ thấy output của transaction chưa commit hoặc đã abort, và mọi lập luận atomic ở trên mất hiệu lực.

transaction.timeout.ms mặc định 60 giây. Xử lý một batch lớn trong một transaction dễ chạm trần này; coordinator abort, producer nhận lỗi và phải xử lý lại. Giả định mỗi record cần 20 ms để gọi payment và 500 record mỗi batch: 10 giây, còn dư. Nếu mỗi record cần 200 ms, batch đó vượt 60 giây và transaction không bao giờ commit được — phải giảm batch, tăng timeout có kiểm soát, hoặc tách xử lý ra khỏi transaction. Broker có transaction.max.timeout.ms làm trần cấu hình, nhưng giá trị cụ thể cần kiểm tra theo phiên bản broker trước khi đưa vào runbook.

Coordinator cũng là một failure domain. Khi coordinator không sẵn sàng, producer không thể begin, commit hay abort; transaction treo tới khi timeout. Ứng dụng phải dừng xử lý và chờ, không được retry mù, vì transaction có thể đã commit ở broker trong khi client chưa nhận được xác nhận.

Lỗi transaction cần phân loại trước khi retry. ProducerFencedException, OutOfOrderSequenceException và AuthorizationException là fatal: phải đóng producer và điều tra, không retry mù. InvalidProducerEpochException thì khác: javadoc xếp nó vào nhóm recoverable, ứng dụng nên abort transaction để producer tự xin epoch mới rồi xử lý lại. Phần lớn lỗi còn lại cũng có thể abort rồi thử transaction mới. commitTransaction() có một trường hợp khó: timeout không có nghĩa là commit thất bại. Javadoc 4.3 nói rõ request có thể đã tới broker; retry commit là an toàn, nhưng không được chuyển sang abort vì commit có thể đang hoàn tất.

Khác biệt 3.x. Từ Kafka 4.0, acks=all và enable.idempotence=true là mặc định, linger.ms tăng từ 0 lên 5. Transaction protocol v2 (KIP-890) bật mặc định trên broker 4.x: epoch được bump mỗi transaction, server tự thêm partition và verify, EndTxn chạy bất đồng bộ. Đổi lại, CONCURRENT_TRANSACTIONS có thể làm produce latency tăng; backoff được xử lý ở server với config riêng.

KIP-939 mở thêm một cửa ra ngoài Kafka: transaction.two.phase.commit.enable=true cho broker biết client tham gia giao thức 2PC, và transaction không hết hạn. Đây là cách để một coordinator bên ngoài commit atomically giữa Kafka và một hệ khác. Cái giá là coordinator phải đáng tin, transaction giữ lock lâu, và toàn bộ hệ phải hiểu giao thức — không phải một config bật là xong. Rủi ro vận hành cụ thể: nếu coordinator ngoài không phản hồi, transaction có thể bị giữ mở, LSO không tiến và consumer read_committed bị chặn. 2PC chỉ phù hợp khi hệ ngoài thực sự tham gia giao thức và có recovery cho coordinator; outbox/CDC thường ít rủi ro hơn cho dual-write.

EOS vì vậy có phạm vi rất cụ thể:

Một lỗi thiết kế hay gặp: ghi DB trước, rồi commit transaction Kafka. Nếu transaction abort, offset không commit nhưng DB đã đổi; consumer xử lý lại và tạo duplicate ở DB. EOS không sửa được thứ tự side effect nằm ngoài Kafka. Muốn atomic giữa DB và event, phải đưa DB vào cùng giao thức (2PC) hoặc đảo chiều bằng outbox.

// props ở C1, thêm transactional.id cho producer này.
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
    "orders-enrichment-1");
producer.initTransactions(); // fence instance cũ, cấp epoch mới

while (running) {
    ConsumerRecords<String, String> records =
        consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> record : records) {
            producer.send(new ProducerRecord<>(
                "orders.enriched",
                record.key(),
                enrich(record.value())));
        }
        // offset và output cùng một transaction.
        // Consumer không tự commit offset.
        producer.sendOffsetsToTransaction(
            offsetsOf(records),
            consumer.groupMetadata());
    } catch (KafkaException e) {
        producer.abortTransaction(); // lỗi trước commit
        handle(e);
        continue;
    }
    commitWithRetry(producer, 3); // commit timeout: retry
}

static void commitWithRetry(
    KafkaProducer<?, ?> producer, int attempts) {
    for (int i = 0; ; i++) {
        try {
            producer.commitTransaction();
            return;
        } catch (TimeoutException e) {
            // Không rõ đã commit: chỉ retry commit, không abort.
            if (i >= attempts) throw e;
        } catch (InvalidProducerEpochException e) {
            // Recoverable: abort để xin epoch mới.
            producer.abortTransaction();
            throw e;
        } catch (ProducerFencedException
                 | OutOfOrderSequenceException
                 | AuthorizationException e) {
            // Fatal: đóng producer và điều tra.
            producer.close(Duration.ZERO);
            throw e;
        } catch (KafkaException e) {
            // Abortable: transaction chưa commit.
            producer.abortTransaction();
            throw e;
        }
    }
}

Ba thiết kế giao nhận, ba failure behavior

At-least-once với sink idempotent là mặc định thực tế của phần lớn hệ thống. Producer 4.x đã durable; consumer commit sau khi xử lý; sink dùng unique key hoặc upsert để chịu duplicate. Không mất nếu commit đúng thứ tự; có thể trùng; không cần coordinator. Điều kiện để nó đúng là sink thực sự idempotent, không chỉ "đã có unique index" mà còn phải xử lý được race giữa hai lần ghi đồng thời.

Transactional read-process-write gộp output Kafka và offset commit vào một transaction. Không trùng trong biên Kafka, nhưng phụ thuộc coordinator, chịu LSO và transaction timeout. Nó đáng dùng khi output là Kafka và invariant cần atomic giữa nhiều topic hoặc giữa output với offset. Nó không làm DB trở nên atomic.

Outbox/CDC ghi DB và outbox trong cùng transaction DB; một connector đọc outbox và publish ra Kafka. Atomic giữa business data và outbox trong DB; publish ra Kafka vẫn có thể lặp và cần retry/idempotent. Latency cao hơn, cần dọn outbox, và phải giữ ordering theo key ở phía publish. Đây là lựa chọn đúng khi DB là source of truth và event chỉ là projection.

Lựa chọn "nhàm chán" đáng cân nhắc: một writer duy nhất cho mỗi key, cộng unique constraint ở DB. Nó không mở rộng ngang tốt, nhưng failure behavior dễ hiểu và không cần coordinator. Với nhiều hệ thống, đây là lựa chọn đúng hơn transaction Kafka.

Điểm crossover nằm ở invariant, không ở độ phổ biến của công nghệ. Nếu duplicate ở sink có thể dedupe với chi phí thấp hơn chi phí coordinator, chọn at-least-once. Nếu cần atomic giữa hai topic trở lên hoặc giữa output và offset, transaction là lựa chọn rẻ hơn so với tự viết bù trừ. Nếu DB là source of truth và event chỉ là projection, outbox thắng vì không phải đưa Kafka vào transaction của DB. Ba lựa chọn không loại trừ nhau: một hệ lớn thường dùng outbox cho luồng ghi DB, transaction cho luồng enrich trong Kafka, và at-least-once cho các luồng idempotent còn lại.

Điểm chung: "exactly-once" là thuộc tính của toàn pipeline. Chọn guarantee yếu nhất còn thoả invariant nghiệp vụ, rồi bù bằng idempotency ở đúng tầng.

Cùng tên config, khác guarantee giữa các client

Java client 4.3 là reference: acks=all, enable.idempotence=true, linger.ms=5, max.in.flight=5. Các client khác dùng cùng tên config nhưng không cùng default. Port nguyên chuỗi config từ Java sang client khác là cách tạo false confidence.

Java client, Spring Kafka và Kafka Streams chạy trên JVM; librdkafka có binding C/Python/Go/.NET; franz-go và Sarama là Go; kafka-python là Python; KafkaJS là Node.js.

Bảng này chỉ so default idempotence và trạng thái maintenance tại 2026-10-25; nó không đánh giá đầy đủ mức hỗ trợ transaction của từng client. Quy tắc thực dụng: port ý định — acks=all, idempotence, delivery timeout, retry budget — rồi kiểm tra default và release note của đúng client sẽ dùng. Một bài test gửi record, kill broker giữa ack và đếm duplicate rẻ hơn nhiều so với tin vào config copy.

Wrapper cũng che default. Spring Kafka thêm retry, error handler và transaction template của riêng nó; Kafka Streams dùng transactional producer bên dưới khi bật processing.guarantee=exactly_once_v2. Trong cả hai trường hợp, config producer vẫn tồn tại nhưng bị một lớp cấu hình khác ghi đè. Phải đọc default của wrapper, không chỉ default của client.

Trạng thái maintenance là một phần của guarantee: client không còn release sẽ không nhận bản vá protocol hoặc security. KafkaJS với bản npm cuối từ 2023 là ví dụ.

Backpressure và fencing là failure mode vận hành thật

Khi buffer cạn, send() block. Nếu service gọi send() trên request thread, thread đó bị giữ và latency của API tăng trước khi broker kịp quá tải. Tăng buffer.memory chỉ đẩy điểm nghẽn ra sau; nếu ingress vượt capacity thật, cần admission control hoặc giảm tải, không phải thêm RAM.

Retry khuếch đại tải. Một broker chậm làm request timeout, producer retry, broker càng chậm. retry.backoff.ms và retry.backoff.max.ms có jitter để tránh đồng bộ, nhưng retry vẫn là tải thật. Theo dõi record-retry-rate cùng request-latency; retry tăng trước khi delivery.timeout.ms xuất hiện.

Con số khuếch đại không nhỏ. Giả định 1% request timeout và mỗi request được thử lại hai lần: tải tăng thêm 2%, chưa tính retry của tầng ứng dụng. Nếu timeout xảy ra vì broker đã bão hoà, 2% đó lại tạo thêm timeout; vòng lặp chỉ dừng khi delivery.timeout.ms cắt batch hoặc khi có admission control. Vì vậy giảm tải ở ingress thường hiệu quả hơn tăng retries.

Admission control phải nằm trước producer, không phải trong producer. Một token bucket hoặc bounded queue ở tầng API giữ cho send() không block request thread; khi vượt ngưỡng, trả lỗi nhanh để caller quyết định, thay vì để buffer đầy rồi mới phát hiện.

Partial failure có ba dạng thường gặp. Leader election gây lỗi tạm thời, producer retry trong delivery.timeout.ms. Coordinator không sẵn sàng làm transaction treo tới timeout. Fencing xảy ra khi hai instance cùng transactional.id; instance cũ nhận lỗi fatal, và đó là hành vi đúng, không phải bug.

Các ngưỡng này là điểm khởi đầu, phải hiệu chỉnh theo workload. Rollout idempotence và acks=all an toàn về semantics nhưng có thể cần chỉnh max.in.flight/retries và có thể tăng latency; rollout transaction cần transactional.id, consumer read_committed, kế hoạch fencing và một giai đoạn dual-read trước khi cắt. Capacity cần tính theo bộ nhớ buffer nhân số instance, CPU cho compression và lưu lượng cross-AZ khi acks=all.

Năm câu hỏi chốt thiết kế giao nhận

1. Duplicate ở đâu là lỗi nghiệp vụ? Nếu chỉ ở payment hoặc DB, làm sink idempotent. Nếu ở output Kafka, cân nhắc transaction.
2. Có cần atomic giữa nhiều topic hoặc giữa output và offset commit không? Nếu có, transaction là công cụ đúng; nếu không, nó thêm coordinator và LSO mà không mua gì.
3. Có side effect ngoài Kafka không? Nếu có, outbox hoặc idempotency key. 2PC chỉ khi hệ ngoài hỗ trợ và cái giá lock được chấp nhận.
4. Ordering theo key có nằm trong invariant không? Nếu có, giữ idempotence; nếu tắt idempotence, đặt max.in.flight=1 hoặc chấp nhận reorder.
5. Ngân sách p99 và availability của phía đọc là bao nhiêu? Nếu chặt, tránh transaction dài và đặt transaction.timeout.ms theo thời gian xử lý thực tế.

Áp dụng cho luồng orders: consumer thanh toán ghi DB và gọi payment service chọn at-least-once với idempotent sink, vì side effect nằm ngoài Kafka và có thể dedupe bằng orderId. Service enrich đọc orders rồi ghi hai topic orders.enriched và orders.audit chọn transaction, vì cần atomic giữa hai output và offset. Service ghi DB rồi phát event chọn outbox, vì DB là source of truth. Ba lựa chọn trong cùng một hệ thống là bình thường.

Bốn tín hiệu buộc thiết kế lại: LSO lag tăng dù consumer khoẻ; BufferExhaustedException xuất hiện dù đã tăng buffer; duplicate vẫn còn sau khi bật idempotence; ProducerFencedException trong lúc deploy. Mỗi tín hiệu chỉ về một tầng khác nhau — transaction quá dài, ingress vượt capacity, app retry hoặc side effect ngoài Kafka, và hai instance cùng transactional.id.

Không có "config chuẩn" cho mọi hệ thống. Có tiêu chí để chọn, và một ngưỡng để biết lựa chọn hiện tại đã hết đúng.

Kết: guarantee là hợp đồng có phạm vi

send() là enqueue. Guarantee bắt đầu từ ack, và chỉ hoàn chỉnh khi biết nó dừng ở đâu.

Default 4.x — acks=all, idempotence, linger.ms=5 — đã có mức bền mà nhiều hệ thống 3.x phải cấu hình thủ công. Transaction chỉ đáng dùng khi atomicity mua được một invariant cụ thể; nếu không, nó thêm coordinator, timeout và LSO mà không giải quyết được side effect ngoài Kafka.

EOS là một hợp đồng có phạm vi. Nó bao phủ record và offset bên trong Kafka; nó không bao phủ DB, HTTP hay payment. KIP-939 mở rộng phạm vi bằng 2PC, nhưng với cái giá coordinator và lock — một ngoại lệ có kiểm soát, không phải mặc định.

Còn một câu hỏi chưa được trả lời trong bài: sau khi record nằm trong log với guarantee đã chọn, consumer đọc, xử lý và commit offset thế nào để không phá guarantee đó? Đó là nội dung phần 3.

Nguồn tham khảo

• Apache Kafka 4.3 producer config: https://kafka.apache.org/43/generated/producer_config.html
• Apache Kafka 4.3 consumer config: https://kafka.apache.org/43/generated/consumer_config.html
• KafkaProducer javadoc 4.3: https://kafka.apache.org/43/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html
• Transaction protocol 4.3 (KIP-890): https://kafka.apache.org/43/operations/transaction-protocol
• Kafka 4.3 source: RecordAccumulator, Sender, TransactionManager, BufferPool và ProducerBatch trong `clients/src/main/java/org/apache/kafka/clients/producer/internals/`.
• ProducerConfig.java 4.3, cho transaction.two.phase.commit.enable (KIP-939): https://github.com/apache/kafka/blob/4.3/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
• librdkafka configuration: https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md
• Sarama config.go: https://github.com/IBM/sarama/blob/main/config.go
• franz-go pkg/kgo/config.go: https://github.com/twmb/franz-go/blob/master/pkg/kgo/config.go
• kafka-python 3.0.11 producer: https://github.com/dpkp/kafka-python/blob/3.0.11/kafka/producer/kafka.py
• KafkaJS trên npm: https://www.npmjs.com/package/kafkajs