Kafka producer: acks, idempotencia y batching

1. Descripción general

Enviar un mensaje a Kafka requiere una línea de send(). La parte difícil viene con las preguntas que la acompañan: ¿el mensaje realmente se ha guardado?, ¿se duplica al hacer reintentos?, ¿cómo enviar decenas de miles de mensajes por segundo sin sobrecargar el broker?

Este artículo cubre tres grupos de configuración que responden esas preguntas: acks, enable.idempotence y batching (batch.size, linger.ms, compression.type). Después de leerlo, sabrás cómo configurar un productor que sea seguro y rápido, tanto con Kafka client puro como con Spring Boot 3.x, y cómo probarlo.

2. Configuración

Con Java puro, solo necesitas kafka-clients:

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

Con Spring Boot 3.x, usa spring-kafka. La versión es gestionada por Spring Boot, y ya trae kafka-clients:

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

A lo largo del artículo, enviaremos 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: ¿cuándo se considera que un mensaje está enviado?

acks determina cuántos reconocimientos del broker debe esperar el productor antes de considerar que una solicitud es exitosa.

Valor	Significado	Riesgo
acks=0	No espera nada	Pierdes mensajes sin enterarte
acks=1	Espera que el líder escriba	Pierdes mensajes si el líder falla antes de que el seguidor copie
acks=all (-1)	Espera que todas las réplicas en el ISR escriban	Latencia ligeramente mayor

Ejemplo de configuración con client 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());
}

Llamar a .get() convierte el envío en síncrono. Lo hacemos aquí solo para imprimir el resultado. Cuando acks=all, get() solo retorna después de que todas las réplicas en el ISR han escrito:

partition=2 offset=17

Usamos customerId como clave, así todos los pedidos de un mismo cliente van a la misma partición y mantienen el orden.

3.1. acks=all no es suficiente sin min.insync.replicas

acks=all significa “todas las réplicas actualmente en el ISR”. Supón que el ISR se reduce a solo el líder. Entonces acks=all no es diferente a acks=1. Necesitas configurar también en el topic o broker:

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

Con replication-factor=3 y min.insync.replicas=2, cada escritura debe tener al menos 2 copias. Si el ISR cae por debajo de 2, el productor recibe NotEnoughReplicasException en lugar de escribir silenciosamente una única copia.

Nota: No establezca min.insync.replicas igual a replication-factor. Con eso, un simple reinicio de broker hace que el topic rechace todas las escrituras.

4. Idempotencia: reintenta sin duplicar

acks=all conlleva reintento. Considera este escenario: el broker escribió el mensaje pero la respuesta se pierde en el camino. El productor cree que falló y reenvía, quedando el mensaje escrito dos veces.

El productor idempotente resuelve exactamente este escenario. Cuando activas enable.idempotence=true:

• El broker asigna un ID de productor (PID) al productor.
• Cada lote enviado a una partición lleva un número de secuencia que aumenta.
• El broker descarta lotes con secuencias que ya ha visto, y rechaza lotes que salten la secuencia. Así se mantiene el orden aunque haya múltiples solicitudes en vuelo.

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 línea:

• ENABLE_IDEMPOTENCE_CONFIG activa el mecanismo de PID y número de secuencia.
• La idempotencia fuerza acks=all. Si estableces acks=1 junto con enable.idempotence=true explícitamente, el productor lanza ConfigException al inicializar.
• max.in.flight.requests.per.connection debe ser ≤ 5. Dentro de este límite, Kafka sigue garantizando el orden dentro de la partición.
• retries debe ser muy grande. El tiempo real de reintentos está limitado por delivery.timeout.ms (2 minutos por defecto).

Desde Kafka 3.0, acks=all y enable.idempotence=true son el defecto. Aun así, deberías declararlos explícitamente. Si no lo haces, basta que alguien establezca acks=1 para que la idempotencia se desactive silenciosamente.

4.1. Lo que la idempotencia no puede hacer

Esta es la parte que frecuentemente se malentiende:

• Solo evita duplicados dentro de una sesión de productor, por partición. Si la aplicación se reinicia, el productor recibe un PID nuevo. Mensajes reenviados después del reinicio se duplicarán.
• No evita duplicados causados por tu código llamando send() dos veces, por ejemplo cuando haces reintentos manuales en un catch.
• No garantiza escritura atómica en múltiples particiones.

Esos casos requieren transacciones (transactional.id) o consumidores que puedan deduplicar según orderId.

5. Batching: intercambiar latencia por throughput

El productor no envía mensaje a mensaje. send() solo coloca el registro en un buffer en memoria, organizado por partición. Un thread de fondo (sender) agrupa registros en lotes y los envía. Tres parámetros principales:

Parámetro	Defecto	Rol
batch.size	16384 bytes	Tamaño máximo de un lote para una partición
linger.ms	0 (5 desde Kafka 4.0)	Tiempo de espera adicional para llenar el lote
compression.type	none	Comprime el lote: gzip, snappy, lz4, zstd
buffer.memory	32 MB	Total de memoria para todos los lotes en espera

Un lote se envía cuando alcanza batch.size o cuando expira linger.ms, lo que ocurra primero. Con linger.ms=0, los lotes casi siempre se envían inmediatamente, así que típicamente contienen solo unos pocos registros.

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

Esta configuración acepta hasta 10 ms de latencia adicional. A cambio, cada solicitud lleva 64 KB de datos comprimidos. Con datos JSON repetitivos como eventos de pedidos, lz4 o zstd típicamente reducen notablemente el ancho de banda de red y espacio en disco del broker. Comprimir por lotes es mucho más eficiente que comprimir mensaje a mensaje, porque lotes más grandes tienen más patrones para explotar.

Puedes verificarlo con las métricas del productor:

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

Envía 10.000 pedidos seguidos con linger.ms=0, luego repite con linger.ms=10, y verás cómo records-per-request-avg aumenta notablemente:

# 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

Los números exactos dependen de tu máquina y carga, pero la tendencia siempre es así.

Nota: Si el buffer se llena (no por que el consumidor sea lento, sino porque el broker es lento o la red está congestionada), send() bloqueará hasta max.block.ms (60 segundos por defecto) y luego lanzará TimeoutException. No llames send() en el thread que maneja requests HTTP sin considerar esta posibilidad.

Nota: delivery.timeout.ms debe ser ≥ linger.ms + request.timeout.ms. Si no, el productor rechaza inicializar.

6. Configuración en Spring Boot 3.x

Spring Boot tiene propiedades propias para parámetros comunes. Los demás van a travé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

El publicador usa KafkaTemplate. Desde Spring Kafka 3.0, send() retorna 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("Fallo al enviar pedido {}", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Pedido {} -> partición {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

Ten en cuenta tres puntos:

• El callback se ejecuta en el thread sender del productor. No hagas trabajo pesado ni llamadas bloqueantes en whenComplete, pues ralentizará todos los envíos subsecuentes.
• Cuando el callback recibe un error, el productor ya ha reintentado todo dentro del límite de delivery.timeout.ms. No reintentes de nuevo llamando send() en el callback. Esa es una fuente de duplicados que la idempotencia no puede detener.
• Devolver CompletableFuture permite que el llamador decida si esperar o no.

7. Pruebas del comportamiento del productor

kafka-clients proporciona MockProducer para probar sin broker. Para usarlo, extraemos la lógica a una clase que acepte la interfaz 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;
    }
}

Prueba con JUnit 5 y 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 prueba verifica un comportamiento:

• Primera prueba usa autoComplete=true, así que cada send() tiene éxito inmediatamente. history() permite inspeccionar las claves, confirmando que el orden por cliente se mantiene.
• Segunda prueba usa autoComplete=false y luego errorNext(...) para simular un broker rechazando la escritura. Es la forma de verificar que tu código maneja errores.
• Última prueba inicializa ProducerConfig para que Kafka valide la combinación de configuraciones. Una combinación inválida, como acks=1 con idempotencia, hará fallar la prueba con ConfigException en lugar de fallar en producción. Kafka normaliza all a "-1", así que la aserción compara con "-1".

8. Cuándo relajar las restricciones

La configuración “maximalmente segura” no siempre es correcta:

• Métricas, logs, clickstreams: perder algunos registros es aceptable. Usa acks=1, linger.ms alto y compresión zstd para mejor throughput.
• Sistemas que necesitan baja latencia por mensaje (ej: cotizaciones en tiempo real): mantén linger.ms pequeño (0–2 ms). Aquí no esperes lotes grandes.
• Pagos, pedidos: siempre usa acks=all, idempotencia y min.insync.replicas=2. Si necesitas exactitud de lectura–proceso–escritura, considera transacciones.

Nota: Establecer max.in.flight.requests.per.connection=1 “para estar seguro” es innecesario cuando ya tienes idempotencia activa. Solo reduce el throughput.

9. Conclusión

Tres grupos de configuración resuelven tres problemas distintos:

• acks=all combinado con min.insync.replicas garantiza sin pérdida de mensajes.
• enable.idempotence garantiza reintentos sin duplicados y mantiene el orden dentro de la partición. Su alcance es solo una sesión de productor.
• batch.size, linger.ms y compression.type determinan el throughput, a costa de algo de latencia.

Declara explícitamente estos valores en lugar de confiar en los valores por defecto. Valídalos con pruebas para evitar sorpresas en producción. Y no hagas reintentos manuales de send() en tu código cuando el productor ya lo hace por ti.