Producer Kafka: acks, retries e idempotencia correctamente

1. Descripción general

Cuando un producer envía mensajes a Kafka, hay tres preguntas que debes responder: ¿cuándo consideramos el envío como exitoso?, ¿cómo reintentar si hay un error?, y ¿los reintentos pueden duplicar mensajes? Este artículo explica acks, retries y enable.idempotence — tres configuraciones que determinan la confiabilidad e integridad de tu producer. Después de leerlo, sabrás cómo configurar un producer “production-ready” y evitarás los errores comunes que causan pérdida o duplicación de datos.

2. Dependencia necesaria

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

Los ejemplos de este artículo usan el cliente puro de Kafka (KafkaProducer), no Spring Kafka, para ver claramente cómo cada configuración afecta el comportamiento.

3. acks: cuándo el producer considera el envío como exitoso

acks determina cuántos brokers deben confirmar antes de que el producer considere que el mensaje fue enviado.

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: el producer no espera ninguna confirmación. El más rápido, pero puede perder mensajes si el broker se cae justo después de recibirlo — incluso sin enterarse de que se perdió.
• acks=1: solo espera a que el líder de la partición escriba en su log local y envíe ACK. Si el líder falla antes de que el seguidor sincronice, el mensaje se pierde.
• acks=all (o -1): espera a que todas las réplicas en el ISR (in-sync replica set) confirmen. La más segura, pero con latencia más alta.

Nota importante: acks=all es realmente seguro solo cuando viene con min.insync.replicas >= 2 en el broker. Si min.insync.replicas=1, acks=all también solo necesita 1 réplica para confirmar — no es más seguro que acks=1.

3.1. Configuración del broker relacionada

# server.properties o configuración de topic
min.insync.replicas=2

Con replication factor = 3 y min.insync.replicas=2, el sistema puede tolerar 1 broker caído y aún escribir, garantizando que los datos residan en al menos 2 copias antes de enviar ACK.

4. retries: manejo de errores temporales

El producer puede encontrar errores temporales como NotLeaderForPartitionException o pérdida de conectividad de red. retries permite que el producer reintente automáticamente sin necesidad de código de manejo manual.

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: número máximo de reintentos. Desde Kafka 2.1, deberías establecerlo muy alto (o Integer.MAX_VALUE) y usar delivery.timeout.ms para controlar el tiempo total de espera en lugar de contar intentos.
• retry.backoff.ms: tiempo de espera entre reintentos, evitando bombardear el broker.
• delivery.timeout.ms: tiempo máximo total desde que se llama send() hasta recibir el resultado (éxito o error final). Debe ser mayor que request.timeout.ms + linger.ms.

Problema: si no tienes cuidado, el reintento puede causar duplicación de mensajes u orden incorrecto de envío.

4.1. Por qué los reintentos causan duplicación de mensajes

1. Producer envía lote con mensaje A
2. Broker escribe A exitosamente, envía ACK de vuelta
3. ACK se pierde en la red (network timeout)
4. Producer no recibe ACK -> lo considera un error -> reintenta
5. Broker recibe la solicitud nuevamente -> escribe A una segunda vez
=> El tema tiene 2 copias del mensaje A

Esta es la situación clásica: el broker escribió exitosamente pero el producer no lo supo, así que reintentó. La idempotencia en la siguiente sección resuelve correctamente este problema.

4.2. Por qué los reintentos causan orden incorrecto

Si max.in.flight.requests.per.connection > 1, el producer puede enviar múltiples lotes en paralelo. El lote 2 podría ser escrito por el broker antes que el lote 1 si el lote 1 falla y está siendo reintentado.

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

Con esta configuración sin idempotencia, el orden de los mensajes en la partición no está garantizado cuando hay reintentos.

5. Idempotencia: resolver duplicación por reintentos

enable.idempotence=true ayuda al producer a evitar automáticamente la duplicación en reintentos, gracias a que Kafka adjunta Producer ID (PID) y sequence number a cada lote.

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

Cuando activas idempotencia, Kafka automáticamente establece los siguientes valores (no necesitas configurarlos tú, pero deberías saberlo):

acks = all
retries = Integer.MAX_VALUE (o muy alto)
max.in.flight.requests.per.connection <= 5

5.1. Cómo funciona el mecanismo

Cada sesión de producer recibe un PID único del broker. Cada mensaje enviado en una partición tiene un sequence number que incrementa. El broker guarda el último sequence number escrito para cada PID + partición.

Producer envía lote seq=5 para partición 0
Broker: "el sequence que tengo es 4, 5 es el siguiente -> escribo, guardo seq=5"
Producer reintenta (por reintentos) lote seq=5
Broker: "seq=5 ya lo escribí -> ignoro, envío ACK de éxito igual"

Gracias a esto, los reintentos no crean registros duplicados — el broker detecta y elimina duplicados basándose en el sequence number, sin necesidad de que el producer o consumer los maneje manualmente.

Nota: la idempotencia solo garantiza no duplicación dentro de una sesión de producer y en el alcance de una partición. Si el producer se reinicia (se asigna un nuevo PID), el mecanismo de dedup según el PID antiguo ya no se aplica. Para exactly-once en múltiples particiones/temas, necesitas transacciones (transactional.id) — fuera del alcance de la idempotencia simple.

5.2. Ejemplo de configuración de 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());

    // Activa idempotencia -> automáticamente establece acks=all, retries alto, 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("Falló enviar orden {}", orderId, exception);
        } else {
            log.info("Orden {} escrita en partición {} offset {}",
                    orderId, metadata.partition(), metadata.offset());
        }
    });
}

• ENABLE_IDEMPOTENCE_CONFIG = true es la línea más importante — automáticamente trae acks=all con ella.
• El callback en send() te dice el resultado real: si exception != null, significa que reintentó hasta que se agotó delivery.timeout.ms pero seguía fallando — solo entonces necesitas manejar el negocio (log, alerta, guardar en tabla de errores).

6. Verificar el comportamiento con tests

Para ilustrar la deduplicación, simulamos un MessageDeduplicator simple que imita el mecanismo de sequence number a nivel de aplicación (úsalo cuando no puedas activar idempotencia, por ejemplo en sistemas 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; // duplicado, ignora
        }
        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(); // reintento con seq igual
        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();
    }
}

El primer test afirma: el mismo sequence reenviado una segunda vez se ignora — correcto con el espíritu de la idempotencia del producer. El segundo test afirma que la deduplicación solo aplica dentro de una clave/partición, sin afectar otras claves.

7. Tabla de síntesis de configuración por necesidad

Necesidad	acks	enable.idempotence	Riesgo
Log, métrica sin importancia	0	false	Pérdida de mensajes en error, pero máximo throughput
Balance velocidad/seguridad	1	false	Pérdida de mensajes si líder falla antes de sincronización
Datos críticos de negocio (pedidos, pagos)	all	true	Latencia más alta, pero sin pérdida, sin duplicación por reintento

8. Errores comunes

• Estableces acks=all pero olvidas min.insync.replicas: crees que es seguro pero el broker solo necesita 1 ISR para confirmar — no es diferente de acks=1.
• Escribes tu propia lógica de reintento en paralelo con el reintento automático del cliente Kafka: fácil de causar duplicación doble porque ambas capas reintentarán.
• Activas idempotencia pero dejas max.in.flight.requests.per.connection > 5: Kafka rechaza esta configuración al inicializar el producer (ConfigException), porque excede el límite que idempotencia soporta.
• Confundes idempotencia con exactly-once en todo el sistema: idempotencia solo protege producer-a-broker. Si el consumer procesa un mensaje y luego llama a un side-effect (enviar email, restar dinero) que no es idempotente, aún puedes procesar duplicados en el consumer.

Nota: no actives enable.idempotence=true si el sistema corre una versión de Kafka más antigua que 0.11 — esta característica aún no es soportada por el broker y el producer fallará al inicializarse.

9. Conclusión

acks, retries e enable.idempotence son tres piezas inseparables cuando necesitas un producer confiable. acks=all garantiza que los datos se repliquen adecuadamente antes de considerarse como exitoso; retries ayuda a superar errores temporales; e idempotencia garantiza que esos reintentos no creen registros duplicados. Con datos críticos de negocio, deberías usar los tres juntos por defecto — el costo de latencia es mucho menor que el costo de manejar datos incorrectos o duplicados después.