Kafka producer: acks, idempotence và batching

1. Tổng quan

Gửi một message lên Kafka chỉ cần một dòng send(). Phần khó nằm ở các câu hỏi đi kèm: message đã thực sự được lưu chưa, có bị ghi trùng khi producer retry không, và làm sao gửi hàng chục nghìn message mỗi giây mà không làm nghẽn broker?

Bài này đi qua ba nhóm cấu hình trả lời ba câu hỏi đó: acks, enable.idempotence và batching (batch.size, linger.ms, compression.type). Đọc xong, bạn sẽ biết cấu hình một producer vừa an toàn vừa nhanh, bằng Kafka client thuần lẫn Spring Boot 3.x, và biết cách test nó.

2. Thiết lập

Với Java thuần, bạn chỉ cần kafka-clients:

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

Với Spring Boot 3.x, dùng spring-kafka. Phiên bản do Spring Boot quản lý, và nó đã kéo theo kafka-clients:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Xuyên suốt bài, ta gửi sự kiện đơn hàng:

public record Order(String id, String customerId, BigDecimal total) {

    public String toJson() {
        return """
            {"id":"%s","customerId":"%s","total":%s}""".formatted(id, customerId, total);
    }
}

3. acks: khi nào một message được coi là đã gửi?

acks quy định producer phải chờ bao nhiêu xác nhận từ broker trước khi coi một request là thành công.

Giá trị	Ý nghĩa	Rủi ro
acks=0	Không chờ gì cả	Mất message mà không hề biết
acks=1	Chờ leader ghi xong	Mất message nếu leader chết trước khi follower kịp sao chép
acks=all (-1)	Chờ mọi replica trong ISR ghi xong	Latency cao hơn một chút

Ví dụ cấu hình bằng client thuần:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost: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");

try (var producer = new KafkaProducer<String, String>(props)) {
    var order = new Order("o-1001", "c-42", new BigDecimal("250000"));
    RecordMetadata meta = producer
        .send(new ProducerRecord<>("orders", order.customerId(), order.toJson()))
        .get();
    System.out.printf("partition=%d offset=%d%n", meta.partition(), meta.offset());
}

Gọi .get() biến lời gửi thành đồng bộ. Ở đây làm vậy chỉ để in kết quả. Khi acks=all, get() chỉ trả về sau khi mọi replica trong ISR đã ghi xong:

partition=2 offset=17

Ta dùng customerId làm key, nên mọi đơn của cùng một khách đi vào cùng một partition và giữ đúng thứ tự.

3.1. acks=all chưa đủ nếu thiếu min.insync.replicas

acks=all nghĩa là “mọi replica đang trong ISR”. Giả sử ISR co lại chỉ còn leader. Khi đó acks=all không khác gì acks=1. Bạn cần cấu hình thêm ở topic hoặc broker:

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders \
  --partitions 6 --replication-factor 3 \
  --config min.insync.replicas=2

Với replication-factor=3 và min.insync.replicas=2, mỗi lần ghi phải có ít nhất 2 bản sao. Nếu ISR tụt xuống dưới 2, producer nhận lỗi NotEnoughReplicasException thay vì âm thầm ghi một bản duy nhất.

Lưu ý: Đừng đặt min.insync.replicas bằng replication-factor. Khi đó chỉ cần một broker restart là topic từ chối mọi lần ghi.

4. Idempotence: retry mà không ghi trùng

acks=all kéo theo retry. Hãy xét kịch bản sau: broker đã ghi message nhưng response bị mất trên đường về. Producer tưởng gửi thất bại nên gửi lại, và message bị ghi hai lần.

Idempotent producer giải quyết đúng kịch bản này. Khi bật enable.idempotence=true:

• Broker cấp cho producer một producer ID (PID).
• Mỗi batch gửi tới một partition mang một sequence number tăng dần.
• Broker loại bỏ batch có sequence đã thấy, và từ chối batch nhảy cóc sequence. Nhờ vậy thứ tự được giữ kể cả khi có nhiều request đang bay.

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

Ý nghĩa từng dòng:

• ENABLE_IDEMPOTENCE_CONFIG bật cơ chế PID và sequence number.
• Idempotence bắt buộc acks=all. Nếu bạn đặt acks=1 cùng enable.idempotence=true một cách tường minh, producer ném ConfigException ngay khi khởi tạo.
• max.in.flight.requests.per.connection phải ≤ 5. Trong giới hạn này, Kafka vẫn đảm bảo thứ tự trong partition.
• retries để rất lớn. Thời gian retry thực tế được giới hạn bởi delivery.timeout.ms (mặc định 2 phút).

Từ Kafka 3.0, acks=all và enable.idempotence=true đã là mặc định. Dù vậy, bạn nên khai báo tường minh. Nếu không, chỉ cần ai đó đặt acks=1 là idempotence âm thầm bị tắt.

4.1. Idempotence không làm được gì

Đây là phần hay bị hiểu sai:

• Nó chỉ chống trùng trong một phiên producer, trên từng partition. Nếu ứng dụng restart, producer nhận PID mới. Message gửi lại sau restart vẫn bị ghi trùng.
• Nó không chống trùng do code của bạn gọi send() hai lần, ví dụ khi bạn tự retry trong catch.
• Nó không cho ghi nguyên tử lên nhiều partition.

Các trường hợp trên cần transaction (transactional.id) hoặc consumer có khả năng khử trùng theo orderId.

5. Batching: đánh đổi latency lấy throughput

Producer không gửi từng message một. send() chỉ đặt record vào một buffer trong bộ nhớ, tổ chức theo từng partition. Một thread nền (sender) gom các record thành batch rồi gửi đi. Ba tham số chính:

Tham số	Mặc định	Vai trò
batch.size	16384 byte	Kích thước tối đa một batch cho một partition
linger.ms	0 (5 từ Kafka 4.0)	Thời gian chờ thêm để gom đầy batch
compression.type	none	Nén cả batch: gzip, snappy, lz4, zstd
buffer.memory	32 MB	Tổng bộ nhớ cho mọi batch đang chờ

Một batch được gửi khi đầy batch.size hoặc khi hết linger.ms, tùy điều kiện nào đến trước. Với linger.ms=0, batch gần như luôn được gửi ngay, nên thường chỉ chứa vài record.

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

Cấu hình này chấp nhận thêm tối đa khoảng 10 ms latency. Đổi lại, mỗi request mang tới 64 KB dữ liệu đã nén. Với dữ liệu JSON lặp nhiều như sự kiện đơn hàng, lz4 hoặc zstd thường giảm đáng kể băng thông mạng và dung lượng đĩa trên broker. Nén theo batch hiệu quả hơn nhiều so với nén từng message, vì batch càng lớn thì càng nhiều chuỗi lặp để khai thác.

Bạn có thể kiểm chứng qua metric của producer:

producer.metrics().forEach((name, metric) -> {
    if (name.name().equals("batch-size-avg") || name.name().equals("records-per-request-avg")) {
        System.out.println(name.name() + " = " + metric.metricValue());
    }
});

Gửi 10.000 đơn hàng liên tục với linger.ms=0, rồi lặp lại với linger.ms=10, bạn sẽ thấy records-per-request-avg tăng rõ rệt:

# linger.ms=0
batch-size-avg = 1830.4
records-per-request-avg = 21.7
# linger.ms=10
batch-size-avg = 41022.9
records-per-request-avg = 486.2

Con số cụ thể phụ thuộc vào máy và tải, nhưng xu hướng thì luôn như vậy.

Lưu ý: Nếu buffer đầy (consumer chậm thì không liên quan, ở đây là broker chậm hoặc mạng nghẽn), send() sẽ block tối đa max.block.ms (mặc định 60 giây) rồi ném TimeoutException. Đừng gọi send() trong thread xử lý request HTTP mà không tính tới khả năng này.

Lưu ý: delivery.timeout.ms phải ≥ linger.ms + request.timeout.ms. Nếu không, producer từ chối khởi tạo.

6. Cấu hình trong Spring Boot 3.x

Spring Boot có property riêng cho các tham số phổ biến. Những tham số còn lại đi qua properties:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      batch-size: 64KB
      compression-type: lz4
      buffer-memory: 64MB
      properties:
        enable.idempotence: true
        linger.ms: 10
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000

Publisher dùng KafkaTemplate. Từ Spring Kafka 3.0, send() trả về CompletableFuture:

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, String>> publish(Order order) {
        return kafkaTemplate.send("orders", order.customerId(), order.toJson())
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Gửi order {} thất bại", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Order {} -> partition {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

Lưu ý ba điểm:

• Callback chạy trên thread sender của producer. Đừng làm việc nặng hoặc gọi blocking trong whenComplete, vì sẽ làm chậm mọi lần gửi khác.
• Khi callback nhận lỗi, producer đã retry hết trong giới hạn delivery.timeout.ms. Đừng retry thêm bằng tay theo kiểu gọi lại send() ngay. Đây chính là nguồn gây trùng lặp mà idempotence không chặn được.
• Trả về CompletableFuture giúp caller tự quyết định có chờ hay không.

7. Test hành vi producer

kafka-clients cung cấp MockProducer để test mà không cần broker. Để dùng nó, ta tách logic vào một lớp nhận interface Producer:

public class OrderSender {

    private final Producer<String, String> producer;

    public OrderSender(Producer<String, String> producer) {
        this.producer = producer;
    }

    public Future<RecordMetadata> send(Order order) {
        return producer.send(new ProducerRecord<>("orders", order.customerId(), order.toJson()));
    }

    public static Properties reliableConfig(String bootstrap) {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        p.put(ProducerConfig.ACKS_CONFIG, "all");
        p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        p.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        p.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
        p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        return p;
    }
}

Test bằng JUnit 5 và AssertJ:

class OrderSenderTest {

    @Test
    void whenSendingOrders_thenKeyIsCustomerId() {
        var mock = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        sender.send(new Order("o-1", "c-42", new BigDecimal("100")));
        sender.send(new Order("o-2", "c-42", new BigDecimal("200")));

        assertThat(mock.history())
            .hasSize(2)
            .extracting(ProducerRecord::key)
            .containsOnly("c-42");
    }

    @Test
    void whenBrokerFails_thenFutureCompletesExceptionally() {
        var mock = new MockProducer<>(false, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        Future<RecordMetadata> future = sender.send(new Order("o-3", "c-7", BigDecimal.TEN));
        mock.errorNext(new NotEnoughReplicasException("ISR < min.insync.replicas"));

        assertThatThrownBy(future::get)
            .hasCauseInstanceOf(NotEnoughReplicasException.class);
    }

    @Test
    void reliableConfig_isAccepted() {
        var props = OrderSender.reliableConfig("localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        var config = new ProducerConfig(props);

        assertThat(config.getString(ProducerConfig.ACKS_CONFIG)).isEqualTo("-1");
        assertThat(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isTrue();
    }
}

Mỗi test kiểm tra một hành vi:

• Test đầu dùng autoComplete=true, nên mọi send() thành công ngay. history() cho phép kiểm tra key, từ đó xác nhận thứ tự theo khách hàng được giữ.
• Test thứ hai dùng autoComplete=false rồi errorNext(...) để giả lập broker từ chối ghi. Đây là cách kiểm tra code xử lý lỗi của bạn.
• Test cuối khởi tạo ProducerConfig để Kafka tự validate tổ hợp cấu hình. Một tổ hợp sai, như acks=1 cùng idempotence, sẽ làm test fail với ConfigException thay vì nổ lúc deploy. Kafka chuẩn hóa all thành "-1", nên assertion so với "-1".

8. Khi nào nên nới lỏng?

Cấu hình “an toàn tối đa” không phải lúc nào cũng đúng:

• Metrics, log, clickstream: mất vài bản ghi chấp nhận được. Khi đó acks=1, linger.ms cao và nén zstd cho throughput tốt nhất.
• Hệ thống cần latency thấp từng message (ví dụ báo giá thời gian thực): giữ linger.ms nhỏ (0–2 ms). Lúc này đừng kỳ vọng batch lớn.
• Thanh toán, đơn hàng: luôn dùng acks=all, idempotence và min.insync.replicas=2. Nếu cần đọc–xử lý–ghi chính xác một lần, hãy cân nhắc transaction.

Lưu ý: Đặt max.in.flight.requests.per.connection=1 “cho chắc ăn” là không cần thiết khi đã bật idempotence. Nó chỉ làm giảm throughput.

9. Kết luận

Ba nhóm cấu hình giải quyết ba vấn đề khác nhau:

• acks=all kết hợp min.insync.replicas đảm bảo không mất message.
• enable.idempotence đảm bảo retry không gây trùng và giữ thứ tự trong partition. Phạm vi của nó chỉ là một phiên producer.
• batch.size, linger.ms và compression.type quyết định throughput, đổi lại một ít latency.

Hãy khai báo tường minh các giá trị này thay vì dựa vào mặc định. Validate chúng bằng test để tránh bất ngờ khi deploy. Và đừng tự retry send() trong code khi producer đã làm việc đó cho bạn.