Kafka producer: acks, idempotence, and batching

1. Overview

Sending a message to Kafka takes just one line of send(). The hard part lies in the questions that follow: has the message actually been persisted, will it be written twice if the producer retries, and how do you send tens of thousands of messages per second without overwhelming the broker?

This article covers three configuration groups that answer those three questions: acks, enable.idempotence, and batching (batch.size, linger.ms, compression.type). By the end, you’ll know how to configure a producer that’s both safe and fast, using either raw Kafka client or Spring Boot 3.x, and how to test it.

2. Setup

With plain Java, you just need kafka-clients:

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

With Spring Boot 3.x, use spring-kafka. The version is managed by Spring Boot, and it already pulls in kafka-clients:

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

Throughout the article, we’ll be sending order events:

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: when is a message considered sent?

acks determines how many confirmations the producer must wait for from the broker before considering a request successful.

Value	Meaning	Risk
acks=0	Don’t wait for anything	Lose messages without knowing
acks=1	Wait for leader to write	Lose messages if leader dies before followers replicate
acks=all (-1)	Wait for all replicas in ISR to write	Slightly higher latency

Example configuration with plain client:

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());
}

Calling .get() makes sending synchronous. Here we do this just to print the result. When acks=all, get() only returns after all replicas in the ISR have written:

partition=2 offset=17

We use customerId as the key, so all orders from the same customer go into the same partition and maintain the correct order.

3.1. acks=all is not enough without min.insync.replicas

acks=all means “all replicas currently in the ISR”. Suppose the ISR shrinks to only the leader. Then acks=all is no different from acks=1. You need to configure it at the topic or broker level:

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

With replication-factor=3 and min.insync.replicas=2, each write must have at least 2 replicas. If the ISR falls below 2, the producer gets a NotEnoughReplicasException instead of silently writing a single copy.

Note: Don’t set min.insync.replicas equal to replication-factor. Then just one broker restart will cause the topic to reject all writes.

4. Idempotence: retry without duplicates

acks=all implies retries. Consider this scenario: the broker writes the message but the response is lost in transit. The producer thinks sending failed so it resends, and the message gets written twice.

An idempotent producer solves this scenario correctly. When you enable enable.idempotence=true:

• The broker assigns the producer a producer ID (PID).
• Each batch sent to a partition carries an incrementing sequence number.
• The broker discards batches with sequences already seen and rejects batches that skip sequences. This way order is preserved even when multiple requests are in flight.

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);

Meaning of each line:

• ENABLE_IDEMPOTENCE_CONFIG enables the PID and sequence number mechanism.
• Idempotence requires acks=all. If you set acks=1 along with enable.idempotence=true explicitly, the producer throws ConfigException at initialization.
• max.in.flight.requests.per.connection must be ≤ 5. Within this limit, Kafka still guarantees ordering within a partition.
• retries should be very large. The actual retry time is limited by delivery.timeout.ms (default 2 minutes).

Since Kafka 3.0, acks=all and enable.idempotence=true are defaults. Still, you should declare them explicitly. Otherwise, someone setting acks=1 will silently disable idempotence.

4.1. What idempotence doesn’t do

This is often misunderstood:

• It only prevents duplicates within a single producer session, per partition. If the application restarts, the producer gets a new PID. Messages resent after restart still get written twice.
• It doesn’t prevent duplicates caused by your code calling send() twice, for example when you retry manually in catch.
• It doesn’t enable atomic writes across multiple partitions.

Those cases require transactions (transactional.id) or a consumer capable of deduplication by orderId.

5. Batching: trading latency for throughput

The producer doesn’t send messages one by one. send() just places records into an in-memory buffer, organized by partition. A background thread (sender) groups records into batches and sends them. Three main parameters:

Parameter	Default	Role
batch.size	16384 bytes	Maximum size of a batch for a partition
linger.ms	0 (5 from Kafka 4.0)	Time to wait before sending a partial batch
compression.type	none	Compress the batch: gzip, snappy, lz4, zstd
buffer.memory	32 MB	Total memory for all pending batches

A batch is sent when batch.size is full or linger.ms expires, whichever comes first. With linger.ms=0, batches are sent almost immediately, so they usually contain just a few records.

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

This configuration accepts up to ~10 ms of added latency. In return, each request carries up to 64 KB of compressed data. With repetitive JSON data like order events, lz4 or zstd often significantly reduces network bandwidth and disk usage on the broker. Batch compression is much more efficient than per-message compression, because larger batches have more repetitive patterns to exploit.

You can verify this through producer metrics:

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());
    }
});

Send 10,000 orders continuously with linger.ms=0, then repeat with linger.ms=10, and you’ll see records-per-request-avg increase noticeably:

# 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

The exact numbers depend on the machine and load, but the trend is always the same.

Note: If the buffer fills up (slow consumer is not relevant here; this is about slow broker or network congestion), send() will block for up to max.block.ms (default 60 seconds) then throw TimeoutException. Don’t call send() from an HTTP request handler thread without accounting for this possibility.

Note: delivery.timeout.ms must be ≥ linger.ms + request.timeout.ms. Otherwise, the producer refuses to initialize.

6. Configuration in Spring Boot 3.x

Spring Boot has dedicated properties for common parameters. Other parameters go through 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 uses KafkaTemplate. Since Spring Kafka 3.0, send() returns a 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("Failed to send order {}", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Order {} -> partition {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

Note three points:

• The callback runs on the producer’s sender thread. Don’t do heavy work or call blocking operations in whenComplete, as it will slow down all other sends.
• When the callback receives an error, the producer has exhausted retries within delivery.timeout.ms. Don’t retry manually by calling send() again immediately. This is exactly the source of duplicates that idempotence can’t prevent.
• Returning CompletableFuture lets the caller decide whether to wait or not.

7. Testing producer behavior

kafka-clients provides MockProducer to test without a broker. To use it, extract logic into a class that accepts a Producer interface:

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 with JUnit 5 and 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();
    }
}

Each test checks one behavior:

• First test uses autoComplete=true, so all send() calls succeed immediately. history() lets you check the key, thus confirming that customer ordering is preserved.
• Second test uses autoComplete=false then errorNext(...) to simulate the broker rejecting writes. This is how you test your error-handling code.
• Last test instantiates ProducerConfig so Kafka validates the configuration combination itself. An invalid combination, like acks=1 with idempotence, causes the test to fail with ConfigException instead of exploding at deployment. Kafka normalizes all to "-1", so assertions compare against "-1".

8. When to relax the settings?

The “maximum safety” configuration isn’t always right:

• Metrics, logs, clickstream: losing a few records is acceptable. Then use acks=1, high linger.ms, and zstd compression for best throughput.
• Systems requiring low per-message latency (e.g., real-time quotes): keep linger.ms small (0–2 ms). Don’t expect large batches then.
• Payments, orders: always use acks=all, idempotence, and min.insync.replicas=2. If you need read–process–write exactly-once, consider transactions.

Note: Setting max.in.flight.requests.per.connection=1 “just to be safe” is unnecessary with idempotence enabled. It only reduces throughput.

9. Conclusion

Three configuration groups solve three different problems:

• acks=all combined with min.insync.replicas ensures no message loss.
• enable.idempotence ensures retries don’t cause duplicates and preserves order within a partition. Its scope is limited to a single producer session.
• batch.size, linger.ms, and compression.type determine throughput, trading off a bit of latency.

Declare these values explicitly instead of relying on defaults. Validate them with tests to avoid surprises at deployment. And don’t retry send() manually in your code when the producer already does it for you.