Kafka producer: acks, idempotência e batching

1. Visão geral

Enviar uma mensagem para o Kafka requer apenas uma linha send(). A parte difícil está nas perguntas que vêm junto: a mensagem foi realmente armazenada, ela não foi escrita em duplicata quando o producer faz retry, e como enviar dezenas de milhares de mensagens por segundo sem sobrecarregar o broker?

Este artigo passa por três grupos de configurações que respondem essas três perguntas: acks, enable.idempotence e batching (batch.size, linger.ms, compression.type). Depois de ler, você saberá como configurar um producer que é tanto seguro quanto rápido, usando o cliente Kafka puro e Spring Boot 3.x, e saberá como testá-lo.

2. Configuração

Com Java puro, você só precisa de kafka-clients:

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

Com Spring Boot 3.x, use spring-kafka. A versão é gerenciada pelo Spring Boot e já traz kafka-clients:

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

Ao longo do artigo, vamos enviar eventos de pedidos:

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: quando uma mensagem é considerada enviada?

acks define quantas confirmações do broker o producer deve esperar antes de considerar uma requisição bem-sucedida.

Valor	Significado	Risco
acks=0	Não espera nada	Perde mensagens sem saber
acks=1	Espera o leader escrever	Perde mensagens se o leader morre antes do follower replicar
acks=all (-1)	Espera cada replica no ISR escrever	Latência um pouco maior

Exemplo de configuração com cliente puro:

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

Chamar .get() torna o envio síncrono. Aqui fazemos isso apenas para imprimir o resultado. Quando acks=all, get() retorna apenas depois que cada replica no ISR termina de escrever:

partition=2 offset=17

Usamos customerId como chave, então todos os pedidos do mesmo cliente vão para a mesma partição e mantêm a ordem correta.

3.1. acks=all não é suficiente sem min.insync.replicas

acks=all significa “cada replica atualmente no ISR”. Suponha que o ISR encolha para apenas o leader. Nesse caso, acks=all não é diferente de acks=1. Você precisa configurar também no tópico ou broker:

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

Com replication-factor=3 e min.insync.replicas=2, cada escrita deve ter pelo menos 2 cópias. Se o ISR cair abaixo de 2, o producer recebe NotEnoughReplicasException em vez de silenciosamente escrever uma única cópia.

Nota: Não defina min.insync.replicas igual a replication-factor. Caso contrário, apenas um reinício de broker faria o tópico rejeitar todas as escritas.

4. Idempotência: retry sem escrever duplicatas

acks=all leva a retry. Considere este cenário: o broker escreveu a mensagem mas a resposta se perdeu no caminho. O producer pensa que falhou e reescreve, e a mensagem é escrita duas vezes.

Um produtor idempotente resolve isso. Quando você ativa enable.idempotence=true:

• O broker atribui um producer ID (PID) ao producer.
• Cada batch enviado para uma partição carrega um número de sequência que aumenta.
• O broker descarta batches com sequências que já viu e rejeita batches que pulam sequências. Isso mantém a ordem mesmo quando múltiplas requisições estão em voo.

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

Significado de cada linha:

• ENABLE_IDEMPOTENCE_CONFIG ativa o mecanismo de PID e número de sequência.
• Idempotência força acks=all. Se você definir acks=1 junto com enable.idempotence=true explicitamente, o producer lança ConfigException ao ser inicializado.
• max.in.flight.requests.per.connection deve ser ≤ 5. Dentro desse limite, Kafka ainda garante ordem dentro da partição.
• retries deve ser bem alto. O tempo real de retry é limitado por delivery.timeout.ms (padrão 2 minutos).

Do Kafka 3.0 em diante, acks=all e enable.idempotence=true são o padrão. Mesmo assim, você deve declarar explicitamente. Caso contrário, se alguém definir acks=1, a idempotência será silenciosamente desativada.

4.1. O que a idempotência não consegue fazer

Esta é a parte frequentemente mal entendida:

• Ela apenas evita duplicatas em uma única sessão do producer, por partição. Se a aplicação reiniciar, o producer recebe um PID novo. Mensagens reenviadas após reinicialização ainda serão duplicadas.
• Ela não evita duplicatas porque seu código chamou send() duas vezes, por exemplo ao fazer retry manualmente em um catch.
• Ela não garante escrita atômica em múltiplas partições.

Esses casos requerem transações (transactional.id) ou um consumer que consiga remover duplicatas por orderId.

5. Batching: trocar latência por throughput

O producer não envia mensagens uma a uma. send() apenas coloca o registro em um buffer na memória, organizado por partição. Uma thread de fundo (sender) agrupa registros em batches e os envia. Três parâmetros principais:

Parâmetro	Padrão	Função
batch.size	16384 bytes	Tamanho máximo de um batch para uma partição
linger.ms	0 (5 no Kafka 4.0)	Tempo de espera adicional para encher o batch
compression.type	none	Comprimir todo o batch: gzip, snappy, lz4, zstd
buffer.memory	32 MB	Total de memória para todos os batches em espera

Um batch é enviado quando fica cheio em batch.size ou quando expira linger.ms, o que acontecer primeiro. Com linger.ms=0, o batch quase sempre é enviado imediatamente, então geralmente contém apenas alguns registros.

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

Essa configuração aceita até 10 ms de latência adicional. Em troca, cada requisição leva até 64 KB de dados comprimidos. Com dados JSON repetitivos como eventos de pedidos, lz4 ou zstd geralmente reduzem significativamente a largura de banda da rede e o uso de disco no broker. Comprimir por batch é muito mais eficiente do que comprimir mensagens individuais, porque batches maiores têm mais repetição para explorar.

Você pode verificar isso através das métricas do 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());
    }
});

Envie 10.000 pedidos consecutivamente com linger.ms=0, depois repita com linger.ms=10, você verá records-per-request-avg aumentar notavelmente:

# 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

Os números exatos dependem da máquina e carga, mas a tendência é sempre a mesma.

Nota: Se o buffer encher (o consumer estar lento não é relevante aqui, é o broker estar lento ou a rede congestionada), send() bloqueará por até max.block.ms (padrão 60 segundos) e então lançará TimeoutException. Não chame send() dentro de uma thread processando requisições HTTP sem levar isso em conta.

Nota: delivery.timeout.ms deve ser ≥ linger.ms + request.timeout.ms. Se não for, o producer recusa inicializar.

6. Configuração no Spring Boot 3.x

O Spring Boot tem propriedades próprias para parâmetros comuns. Os outros parâmetros vão através de 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

O publisher usa KafkaTemplate. A partir do Spring Kafka 3.0, send() retorna um 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("Falha ao enviar pedido {}", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Pedido {} -> partição {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

Observe três pontos:

• O callback executa em uma thread sender do producer. Não faça trabalho pesado ou bloqueante em whenComplete, pois isso atrasará todos os outros envios.
• Quando o callback recebe um erro, o producer já tentou novamente dentro do limite de delivery.timeout.ms. Não faça retry adicional manualmente chamando send() novamente. Essa é exatamente a origem de duplicatas que idempotência não consegue bloquear.
• Retornar CompletableFuture permite que o chamador decida se quer esperar ou não.

7. Testar o comportamento do producer

kafka-clients fornece MockProducer para testar sem precisar de um broker. Para usá-lo, extraia a lógica em uma classe que aceita uma 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;
    }
}

Teste com JUnit 5 e 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();
    }
}

Cada teste verifica um comportamento:

• Primeiro teste usa autoComplete=true, então cada send() sucede imediatamente. history() permite verificar a chave, confirmando que a ordem por cliente é mantida.
• Segundo teste usa autoComplete=false e depois errorNext(...) para simular o broker rejeitando uma escrita. Essa é a forma de testar como seu código lida com erros.
• Último teste inicializa ProducerConfig para validar a combinação de configurações do Kafka. Uma combinação inválida, como acks=1 com idempotência, faz o teste falhar com ConfigException em vez de explodir durante a implantação. Kafka normaliza all para "-1", então a assertion compara com "-1".

8. Quando afrouxar a configuração?

A configuração “máxima segurança” nem sempre é a correta:

• Métricas, logs, clickstream: perder alguns registros é aceitável. Nesse caso, acks=1, linger.ms alto e compressão zstd para melhor throughput.
• Sistemas que precisam de latência baixa por mensagem (por exemplo, cotações em tempo real): mantenha linger.ms pequeno (0–2 ms). Nessa situação, não espere batches grandes.
• Pagamentos, pedidos: sempre use acks=all, idempotência e min.insync.replicas=2. Se precisar ler–processar–escrever exatamente uma vez, considere transações.

Nota: Definir max.in.flight.requests.per.connection=1 “para estar seguro” é desnecessário quando idempotência está ativada. Isso apenas reduz o throughput.

9. Conclusão

Três grupos de configurações resolvem três problemas diferentes:

• acks=all combinado com min.insync.replicas garante não perder mensagens.
• enable.idempotence garante que retry não cause duplicatas e mantém ordem dentro da partição. Seu escopo é apenas uma sessão do producer.
• batch.size, linger.ms e compression.type determinam o throughput, em troca de um pouco de latência.

Declare explicitamente esses valores em vez de confiar nos padrões. Valide-os com testes para evitar surpresas na implantação. E não faça retry de send() no seu código quando o producer já faz isso por você.