Productores y consumidores de Kafka en Java: offsets, idempotencia y transacciones

El cliente Java de Kafka esconde mucha maquinaria detrás de dos clases. KafkaProducer.send parece una simple llamada a un método, pero es la entrada de un pipeline asíncrono de batching con reintentos, y KafkaConsumer.poll parece decir «dame mensajes», pero además gestiona los heartbeats, la pertenencia al grupo y los offsets. Conseguir un comportamiento fiable pasa por entender qué hacen esas dos llamadas y elegir a conciencia un puñado de ajustes. Esta guía usa org.apache.kafka:kafka-clients 3.x; la API es estable en todas las versiones 3.x y 4.x.

El pipeline del productor

Cuando llamas a send, el registro se serializa, se asigna a una partición y se añade a un batch en memoria de esa partición. Un thread de I/O en segundo plano envía los batches llenos —o los que llevan más de linger.ms esperando— a los líderes de las particiones, espera los acknowledgements según acks, reintenta los fallos y, por último, completa el Future e invoca tu callback. Cuando send retorna, todavía no ha salido nada del proceso.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092,kafka-3:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.CLIENT_ID_CONFIG, "order-service");

// Durability: wait for the in-sync replicas, never reorder, never duplicate on retry
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);   // total budget for retries

// Throughput: batch for up to 20 ms, compress batches
props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

try (Producer<String, String> producer = new KafkaProducer<>(props)) {
  ProducerRecord<String, String> record =
      new ProducerRecord<>("orders", "customer-4711", "{\"orderId\":\"o-9911\",\"total\":149.90}");
  record.headers().add("schemaVersion", "2".getBytes(StandardCharsets.UTF_8));

  producer.send(record, (metadata, exception) -> {
    if (exception != null) {
      log.error("failed to send order o-9911", exception);         // after all retries were exhausted
    } else {
      log.debug("o-9911 → {}-{} @ {}", metadata.topic(), metadata.partition(), metadata.offset());
    }
  });
  // close() flushes pending batches and waits for their callbacks
}

Los ajustes que importan

La idempotencia es el ajuste clave. Sin ella, un batch que el líder escribió pero cuyo acknowledgement se perdió se reintenta y se escribe dos veces, y un batch anterior reintentado puede acabar detrás de uno posterior. Con ella, el broker reconoce el reintento por su número de secuencia y conserva exactamente una copia, en orden. No tiene un coste medible y viene activada por defecto con acks=all, pero cualquier acks=1 o acks=0 explícito la desactiva sin avisar, lo que suele explicar la aparición de duplicados tras una actualización.

send también lanza excepciones de forma síncrona en algunos casos: fallo de serialización, un registro demasiado grande para max.request.size y un buffer lleno tras max.block.ms. Gestiona tanto la excepción como el callback.

El bucle de poll del consumidor

Un consumidor se suscribe a topics y después llama a poll en un bucle. Cada poll devuelve un lote de registros de las particiones asignadas a ese consumidor y, como efecto secundario, envía heartbeats, se une o vuelve a unirse al grupo y, con auto-commit, hace commit de los offsets. Ese efecto secundario es la razón por la que el bucle debe seguir llamando a poll aunque no haya nada que procesar.

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092,kafka-3:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");  // first start: read from the beginning
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);     // we commit after processing
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300_000);

try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
  consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
      consumer.commitSync();          // commit what we finished before losing the partitions
    }
    @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { }
  });

  while (running.get()) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
      billing.process(record.key(), record.value());       // must be idempotent: redelivery is possible
    }
    consumer.commitSync();            // after the whole batch: at-least-once
  }
} catch (WakeupException ignored) {
  // consumer.wakeup() was called from a shutdown hook
}

auto.offset.reset solo se aplica cuando el grupo no tiene ningún offset confirmado para una partición: earliest procesa el histórico y latest (el valor por defecto) empieza solo por los registros nuevos. Una vez que el grupo ha hecho commit, el ajuste deja de tener efecto, lo que sorprende a quien lo cambia y no ve ninguna diferencia.

max.poll.records limita cuánto trabajo abarca una iteración, y max.poll.interval.ms cuánto puede durar. Si procesar 200 registros puede superar los cinco minutos en el peor caso, baja el primero o sube el segundo, o bien lleva el trabajo a otro thread y llama a pause sobre las particiones mientras se ejecuta, para que poll pueda mantener viva la pertenencia al grupo.

Detén un consumidor desde otro thread con consumer.wakeup(), que hace que el poll bloqueado lance una WakeupException; después cierra el consumidor en su propio thread. Cerrar no hace commit de nada por sí solo salvo que el auto-commit esté activado, así que la ruta de apagado debe hacer commit explícitamente.

Hacer commit de los offsets

El offset que confirmas es la posición del siguiente registro que se va a leer. El punto del bucle en el que haces commit determina el tamaño de la ventana de duplicados o de pérdida tras una caída.

Un patrón habitual combina las dos formas: commitAsync dentro del bucle por rendimiento y commitSync en el bloque finally y en onPartitionsRevoked por corrección. Elijas lo que elijas, el consumidor debe tolerar ver un registro dos veces —una caída entre el procesamiento y el commit no se puede evitar—, así que el procesamiento debe ser idempotente, normalmente guardando (topic, partición, offset) o una clave de negocio junto al resultado.

Si los offsets encajan mejor junto a los datos —en la misma transacción de base de datos que la fila que generó el registro—, desactiva el auto-commit, escribe el offset en la base de datos y, al arrancar, léelo y haz seek de cada partición asignada hasta él desde onPartitionsAssigned. Así obtienes exactly-once entre Kafka y esa base de datos sin transacciones de Kafka.

Transacciones y exactly-once

En los pipelines que leen de Kafka, procesan y vuelven a escribir en Kafka, las transacciones hacen atómicos los registros de salida y el commit de los offsets de entrada: o las salidas son visibles y los offsets quedan confirmados, o no ocurre ninguna de las dos cosas. Junto con la idempotencia, esto es la semántica exactly-once (EOS) de Kafka.

Properties pp = producerProps();
pp.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "billing-enricher-" + instanceId);  // stable per instance
Producer<String, String> producer = new KafkaProducer<>(pp);
producer.initTransactions();

Properties cp = consumerProps();
cp.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");   // never see aborted records
cp.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
Consumer<String, String> consumer = new KafkaConsumer<>(cp);
consumer.subscribe(List.of("orders"));

while (running.get()) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
  if (records.isEmpty()) continue;
  producer.beginTransaction();
  try {
    for (ConsumerRecord<String, String> r : records) {
      producer.send(new ProducerRecord<>("invoices", r.key(), enrich(r.value())));
    }
    // Offsets are committed as part of the transaction, not by the consumer
    producer.sendOffsetsToTransaction(nextOffsets(records), consumer.groupMetadata());
    producer.commitTransaction();
  } catch (ProducerFencedException | OutOfOrderSequenceException e) {
    throw e;                              // another instance took over this transactional id; stop
  } catch (KafkaException e) {
    producer.abortTransaction();          // inputs will be re-polled and reprocessed
  }
}

transactional.id identifica al productor entre reinicios: una nueva instancia con el mismo id aísla (fences) a la anterior, de modo que una instancia zombi no puede confirmar trabajo obsoleto. Los consumidores downstream solo ven las salidas si usan read_committed; con el valor por defecto, read_uncommitted, también verían los registros de transacciones abortadas. Las transacciones cuestan latencia —un commit supone un viaje de ida y vuelta al coordinador de transacciones más marcadores en cada partición escrita— y conviene usarlas con lotes de registros, no registro a registro.

Kafka Streams activa todo esto con un único ajuste (processing.guarantee=exactly_once_v2), que suele ser el camino más sencillo cuando todo el pipeline va de Kafka a Kafka.

Serialización

El cliente incluye serializadores para strings, enteros, longs, arrays de bytes y algunos tipos más. Los payloads reales usan JSON con esquema, o Avro o Protocol Buffers a través de un schema registry, cuyos serializadores anteponen un identificador de esquema a cada registro para que los consumidores puedan evolucionar de forma independiente. Sea cual sea el formato, un fallo de deserialización en el consumidor es fatal para el bucle de poll si no se gestiona: el ErrorHandlingDeserializer de Spring Kafka, o un wrapper que devuelva un valor centinela, permite al bucle saltarse un registro corrupto y continuar en lugar de fallar con él una y otra vez.

Spring Kafka

Spring for Apache Kafka envuelve el mismo cliente. KafkaTemplate.send devuelve un CompletableFuture sobre el mismo productor; @KafkaListener ejecuta un bucle de poll por cada thread del contenedor, convierte los registros y hace commit según un AckMode (BATCH tras cada poll por defecto, MANUAL cuando tú mismo llamas a Acknowledgment.acknowledge(), RECORD tras cada registro).

@Component
class OrderListener {

  @KafkaListener(topics = "orders", groupId = "billing", concurrency = "3")
  public void onOrder(ConsumerRecord<String, String> record, Acknowledgment ack) {
    billing.process(record.key(), record.value());
    ack.acknowledge();                                  // AckMode.MANUAL: commit after this record's batch
  }
}

concurrency = "3" crea tres threads consumidores en el grupo, cada uno con su propio KafkaConsumer, así que no debería superar el número de particiones. Las propiedades spring.kafka.* de Boot configuran los serializadores, acks, el group id y el modo de acknowledgement del listener, y un DefaultErrorHandler con un DeadLetterPublishingRecoverer te da el comportamiento de «reintentar y luego enviar a un dead-letter topic» que Kafka puro no ofrece:

@Bean
DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
  var recoverer = new DeadLetterPublishingRecoverer(template);            // → orders.DLT by default
  return new DefaultErrorHandler(recoverer, new FixedBackOff(1_000L, 3)); // 3 retries, 1 s apart
}

Como un registro fallido bloquea su partición mientras se reintenta, mantén el backoff corto y lleva lo que necesite esperas largas a un topic de reintentos aparte; @RetryableTopic de Spring Kafka automatiza ese patrón.

Errores comunes

• acks=1 o acks=0 «por rendimiento». Desactiva la idempotencia y acepta pérdidas si cae el líder. El batching y la compresión dan mucho más rendimiento sin esa pérdida.
• Fire-and-forget sin callback. send retorna antes de que ocurra nada; un registro fallido desaparece si no se comprueba el callback o el Future.
• Procesar en otro thread con el auto-commit activado. El siguiente poll confirma registros que ese thread no ha terminado. Haz commit manual tras procesar, o pausa las particiones.
• Consumidores cuyo efecto secundario no es idempotente. At-least-once significa que habrá duplicados: tras un rebalanceo, una caída o un reintento. Elimina duplicados por clave o por offset.
• Bloquear dentro del bucle de poll más tiempo que max.poll.interval.ms. El consumidor es expulsado, el grupo se rebalancea y el commit acaba fallando. Limita el trabajo por poll.
• Usar commitAsync en todas partes. Sus commits no se reintentan, así que un fallo transitorio durante el apagado hace perder el progreso. Termina con commitSync.
• Compartir un KafkaConsumer entre threads. El cliente lo detecta y lanza ConcurrentModificationException; usa un consumidor por thread.

Ideas clave

• send añade a un batch; un thread de I/O envía, reintenta y confirma. Mantén acks=all y la idempotencia, ajusta linger.ms, batch.size y la compresión para el rendimiento, y comprueba siempre el callback.
• poll es el latido del consumidor: sigue llamándolo, limita el trabajo por llamada con max.poll.records y no superes max.poll.interval.ms.
• El offset confirmado es el siguiente registro que se leerá; haz commit después de procesar para tener at-least-once y haz que el procesamiento sea idempotente, porque los duplicados tras una caída o un rebalanceo son inevitables.
• Usa commitAsync en el bucle por velocidad y commitSync en la revocación y el apagado; guarda los offsets en la base de datos de destino cuando necesites exactly-once hacia un sistema que no sea Kafka.
• Las transacciones con transactional.id, sendOffsetsToTransaction y consumidores en read_committed dan exactly-once en pipelines de Kafka a Kafka; Kafka Streams ofrece lo mismo ya empaquetado.
• KafkaTemplate, @KafkaListener, DefaultErrorHandler y el dead-letter recoverer de Spring Kafka envuelven el cliente sin cambiar ninguna de estas reglas.