Producer Kafka: acks, retry và idempotence đúng cách

1. Tổng quan

Khi producer gửi message lên Kafka, có ba câu hỏi bạn phải trả lời: gửi xong chưa coi là thành công, lỡ lỗi thì retry thế nào, và retry có làm trùng message không. Bài này giải thích acks, retries và enable.idempotence — ba cấu hình quyết định độ tin cậy và tính đúng đắn của producer. Đọc xong, bạn sẽ biết cấu hình producer “production-ready” và tránh được các lỗi phổ biến gây mất hoặc trùng dữ liệu.

2. Dependency cần thiết

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

Ví dụ trong bài dùng Kafka client thuần (KafkaProducer), không qua Spring Kafka, để thấy rõ từng config ảnh hưởng thế nào.

3. acks: khi nào producer coi là gửi thành công

acks quyết định producer phải chờ bao nhiêu broker xác nhận trước khi coi message đã gửi xong.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");

• acks=0: producer không chờ xác nhận gì cả. Nhanh nhất, nhưng mất message nếu broker down ngay sau khi nhận — thậm chí không biết là mất.
• acks=1: chỉ chờ leader partition ghi vào log local rồi trả ACK. Nếu leader crash trước khi follower kịp sync, message mất.
• acks=all (hoặc -1): chờ tất cả replica trong ISR (in-sync replica set) xác nhận. An toàn nhất, nhưng latency cao hơn.

Lưu ý: acks=all chỉ thật sự an toàn khi đi kèm min.insync.replicas >= 2 ở broker. Nếu min.insync.replicas=1, acks=all cũng chỉ cần 1 replica xác nhận — không khác gì acks=1 về độ an toàn.

3.1. Cấu hình broker đi kèm

# server.properties hoặc topic config
min.insync.replicas=2

Với replication factor = 3 và min.insync.replicas=2, hệ thống chịu được 1 broker chết mà vẫn ghi được, đồng thời đảm bảo dữ liệu nằm trên ít nhất 2 bản sao trước khi ACK.

4. retries: xử lý lỗi tạm thời

Producer có thể gặp lỗi tạm thời như NotLeaderForPartitionException hay mất kết nối mạng ngắn. retries cho phép producer tự gửi lại mà không cần code xử lý thủ công.

props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

• retries: số lần retry tối đa. Từ Kafka 2.1, nên set rất cao (hoặc Integer.MAX_VALUE) và dùng delivery.timeout.ms để kiểm soát tổng thời gian chờ thay vì đếm số lần.
• retry.backoff.ms: thời gian chờ giữa các lần retry, tránh spam broker.
• delivery.timeout.ms: tổng thời gian tối đa từ lúc gọi send() đến lúc nhận kết quả (thành công hoặc lỗi cuối cùng). Phải lớn hơn request.timeout.ms + linger.ms.

Vấn đề: nếu không cẩn thận, retry có thể làm message bị gửi trùng hoặc gửi sai thứ tự.

4.1. Vì sao retry gây trùng message

1. Producer gửi batch chứa message A
2. Broker ghi A thành công, gửi ACK về
3. ACK bị mất trên đường mạng (network timeout)
4. Producer không nhận được ACK -> coi là lỗi -> retry
5. Broker nhận lại request -> ghi A lần 2
=> Topic có 2 bản của message A

Đây là tình huống kinh điển: broker đã ghi thành công nhưng producer không biết, nên gửi lại. Idempotence ở phần sau giải quyết đúng vấn đề này.

4.2. Vì sao retry gây sai thứ tự

Nếu max.in.flight.requests.per.connection > 1, producer có thể gửi nhiều batch song song. Batch 2 có thể được broker ghi trước batch 1 nếu batch 1 bị lỗi và đang retry.

max.in.flight.requests.per.connection=5

Với cấu hình này mà không có idempotence, thứ tự message trong partition không được đảm bảo khi có retry.

5. Idempotence: giải quyết trùng lặp do retry

enable.idempotence=true giúp producer tự động tránh ghi trùng khi retry, nhờ Kafka gắn Producer ID (PID) và sequence number cho mỗi batch.

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

Khi bật idempotence, Kafka tự động set các giá trị sau (bạn không cần tự cấu hình, nhưng nên biết):

acks = all
retries = Integer.MAX_VALUE (hoặc rất lớn)
max.in.flight.requests.per.connection <= 5

5.1. Cơ chế hoạt động

Mỗi producer session được broker cấp một PID duy nhất. Mỗi message gửi trong một partition có một sequence number tăng dần. Broker lưu sequence number cuối cùng đã ghi cho mỗi PID + partition.

Producer gửi batch seq=5 cho partition 0
Broker: "sequence cuối tôi lưu là 4, 5 là tiếp theo -> ghi, lưu seq=5"
Producer gửi lại (do retry) batch seq=5
Broker: "seq=5 tôi đã ghi rồi -> bỏ qua, trả ACK thành công luôn"

Nhờ vậy, retry không tạo bản ghi trùng — broker tự phát hiện và loại bỏ duplicate dựa trên sequence number, không cần producer hay consumer tự xử lý.

Lưu ý: idempotence chỉ đảm bảo không trùng trong một session producer và trong phạm vi một partition. Nếu producer restart (PID mới được cấp), cơ chế dedup theo PID cũ không còn áp dụng. Muốn exactly-once xuyên suốt nhiều partition/topic, bạn cần transaction (transactional.id) — nằm ngoài phạm vi idempotence đơn thuần.

5.2. Ví dụ cấu hình producer production-ready

public ProducerFactory() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // Bật idempotence -> tự set acks=all, retries cao, in-flight <= 5
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

    this.producer = new KafkaProducer<>(props);
}

public void sendOrderEvent(String orderId, String payload) {
    ProducerRecord<String, String> record =
            new ProducerRecord<>("order-events", orderId, payload);

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("Gửi order {} thất bại", orderId, exception);
        } else {
            log.info("Order {} ghi vào partition {} offset {}",
                    orderId, metadata.partition(), metadata.offset());
        }
    });
}

• ENABLE_IDEMPOTENCE_CONFIG = true là dòng quan trọng nhất — nó kéo theo acks=all tự động.
• Callback trong send() cho biết kết quả thực tế: nếu exception != null, nghĩa là đã retry hết delivery.timeout.ms mà vẫn lỗi — lúc này bạn mới cần xử lý nghiệp vụ (log, alert, lưu vào bảng lỗi).

6. Kiểm tra hành vi với test

Để minh họa phần dedup, ta giả lập một MessageDeduplicator đơn giản mô phỏng cơ chế sequence number ở mức application (dùng khi bạn không thể bật idempotence, ví dụ hệ thống legacy):

public class MessageDeduplicator {

    private final Map<String, Long> lastSeenSequence = new ConcurrentHashMap<>();

    public boolean shouldWrite(String partitionKey, long sequence) {
        Long last = lastSeenSequence.get(partitionKey);
        if (last != null && sequence <= last) {
            return false; // duplicate, bỏ qua
        }
        lastSeenSequence.put(partitionKey, sequence);
        return true;
    }
}

class MessageDeduplicatorTest {

    @Test
    void shouldSkipDuplicateSequence() {
        MessageDeduplicator deduplicator = new MessageDeduplicator();

        assertThat(deduplicator.shouldWrite("order-1", 5)).isTrue();
        assertThat(deduplicator.shouldWrite("order-1", 5)).isFalse(); // retry trùng seq
        assertThat(deduplicator.shouldWrite("order-1", 6)).isTrue();
    }

    @Test
    void shouldAllowDifferentPartitionKeysIndependently() {
        MessageDeduplicator deduplicator = new MessageDeduplicator();

        assertThat(deduplicator.shouldWrite("order-1", 1)).isTrue();
        assertThat(deduplicator.shouldWrite("order-2", 1)).isTrue();
    }
}

Test đầu khẳng định: cùng sequence gửi lại lần hai bị bỏ qua — đúng tinh thần idempotence producer. Test thứ hai khẳng định việc dedup chỉ áp dụng trong phạm vi một key/partition, không ảnh hưởng key khác.

7. Bảng tổng hợp cấu hình theo nhu cầu

Nhu cầu	acks	enable.idempotence	Rủi ro
Log, metric không quan trọng	0	false	Mất message khi lỗi, nhưng throughput cao nhất
Cân bằng tốc độ/an toàn	1	false	Mất message nếu leader crash trước khi follower sync
Dữ liệu nghiệp vụ quan trọng (đơn hàng, thanh toán)	all	true	Latency cao hơn, nhưng không mất, không trùng do retry

8. Lỗi hay gặp

• Set acks=all nhưng quên min.insync.replicas: nghĩ là an toàn nhưng broker chỉ cần 1 ISR vẫn ACK — không khác acks=1.
• Tự viết retry logic song song với retry mặc định của Kafka client: dễ gây gửi trùng gấp đôi vì cả hai lớp retry đều hoạt động.
• Bật idempotence nhưng để max.in.flight.requests.per.connection > 5: Kafka sẽ reject cấu hình này ngay từ khi khởi tạo producer (ConfigException), vì vượt quá giới hạn mà idempotence hỗ trợ.
• Nhầm idempotence với exactly-once toàn hệ thống: idempotence chỉ bảo vệ producer-to-broker. Nếu consumer xử lý message rồi gọi side-effect (gửi email, trừ tiền) không idempotent, bạn vẫn có thể xử lý trùng ở phía consumer.

Lưu ý: không nên bật enable.idempotence=true nếu hệ thống đang chạy Kafka version cũ hơn 0.11 — tính năng này chưa được broker hỗ trợ và producer sẽ lỗi khi khởi tạo.

9. Kết luận

acks, retries và enable.idempotence là ba mảnh ghép không thể tách rời khi bạn cần producer đáng tin cậy. acks=all đảm bảo dữ liệu được replicate đủ trước khi coi là thành công; retries giúp vượt qua lỗi tạm thời; và idempotence đảm bảo việc retry đó không tạo ra bản ghi trùng. Với dữ liệu nghiệp vụ quan trọng, hãy mặc định dùng cả ba cùng nhau — chi phí latency nhỏ hơn rất nhiều so với chi phí xử lý dữ liệu sai hoặc trùng về sau.