Producers e consumers Kafka em Java: offsets, idempotência e transações

O cliente Java do Kafka esconde muita engrenagem atrás de duas classes. KafkaProducer.send parece uma chamada de método comum, mas é a porta de entrada de um pipeline assíncrono de batching com retries, e KafkaConsumer.poll parece só “me dê mensagens”, mas também cuida dos heartbeats, da participação no grupo e do gerenciamento de offsets. Obter um comportamento confiável exige entender o que essas duas chamadas fazem e escolher deliberadamente um punhado de configurações. Este guia usa org.apache.kafka:kafka-clients 3.x; a API é estável em todo o 3.x e o 4.x.

O pipeline do producer

Quando você chama send, o registro é serializado, atribuído a uma partição e adicionado a um batch em memória daquela partição. Uma thread de I/O em segundo plano envia os batches cheios — ou os que já esperam há mais de linger.ms — para os líderes das partições, espera os acknowledgements conforme acks, repete os envios que falharam e, por fim, completa o Future e invoca seu callback. Quando send retorna, nada saiu do processo ainda.

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
}

As configurações que importam

A idempotência é a configuração mais importante. Sem ela, um batch que o líder gravou mas cujo acknowledgement se perdeu é reenviado e gravado duas vezes, e um batch anterior reenviado pode acabar atrás de um posterior. Com ela, o broker reconhece o reenvio pelo número de sequência e mantém exatamente uma cópia, na ordem. O custo não é mensurável, e ela vem habilitada por padrão com acks=all — mas qualquer acks=1 ou acks=0 explícito a desabilita silenciosamente, um motivo comum para duplicatas depois de um upgrade.

send também lança exceções de forma síncrona em algumas situações: falha de serialização, registro grande demais para max.request.size e buffer cheio após max.block.ms. Trate tanto a exceção quanto o callback.

O loop de poll do consumer

Um consumer se inscreve em topics e depois chama poll em um loop. Cada poll retorna um lote de registros das partições atribuídas a esse consumer e, como efeito colateral, envia heartbeats, entra ou volta a entrar no grupo e — com auto-commit — faz commit dos offsets. É por causa desse efeito colateral que o loop precisa continuar chamando poll mesmo quando não há nada para processar.

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 só vale quando o grupo não tem nenhum offset commitado para uma partição: earliest processa o histórico, latest (o padrão) começa apenas pelos registros novos. Depois que o grupo faz commit, a configuração deixa de ter efeito, o que surpreende quem a altera e não vê diferença nenhuma.

max.poll.records limita quanto trabalho uma iteração carrega, e max.poll.interval.ms limita quanto tempo ela pode levar. Se processar 200 registros pode passar de cinco minutos no pior caso, diminua o primeiro ou aumente o segundo, ou então mova o trabalho para outra thread e chame pause nas partições enquanto ele roda, para que poll mantenha a participação no grupo.

Pare um consumer a partir de outra thread com consumer.wakeup(), que faz o poll bloqueado lançar WakeupException; depois feche o consumer na própria thread dele. Fechar não faz commit de nada sozinho, a menos que o auto-commit esteja ligado, então o caminho de desligamento deve fazer commit explicitamente.

Commit de offsets

O offset que você commita é a posição do próximo registro a ser lido. O ponto do loop em que você faz commit define o tamanho da janela de duplicatas ou de perda depois de uma queda.

Um padrão comum combina as duas formas: commitAsync no loop para ganhar throughput e commitSync no bloco finally e em onPartitionsRevoked para garantir a correção. Seja qual for a escolha, o consumer precisa tolerar ver um registro duas vezes — uma queda entre o processamento e o commit não tem como ser evitada —, então o processamento deve ser idempotente, normalmente registrando (topic, partição, offset) ou uma chave de negócio junto com o resultado.

Se faz mais sentido guardar os offsets junto com os dados — na mesma transação de banco que a linha gerada pelo registro —, desabilite o auto-commit, grave o offset no banco e, na inicialização, leia-o de volta e faça seek de cada partição atribuída até ele a partir de onPartitionsAssigned. Isso dá exactly-once entre o Kafka e esse banco sem transações do Kafka.

Transações e exactly-once

Em pipelines que leem do Kafka, processam e gravam de volta no Kafka, as transações tornam atômicos os registros de saída e o commit dos offsets de entrada: ou as saídas ficam visíveis e os offsets são commitados, ou nenhuma das duas coisas acontece. Somada à idempotência, essa é a semântica exactly-once (EOS) do 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 o producer entre reinicializações: uma nova instância com o mesmo id isola (fences) a antiga, de modo que uma instância zumbi não consegue commitar trabalho obsoleto. Os consumers downstream só enxergam as saídas quando usam read_committed; com o padrão read_uncommitted, veriam também registros de transações abortadas. Transações custam latência — um commit é uma ida e volta ao coordenador de transações mais marcadores em cada partição gravada — e funcionam melhor com lotes de registros do que com um registro por vez.

O Kafka Streams habilita tudo isso com uma única configuração (processing.guarantee=exactly_once_v2), que costuma ser o caminho mais fácil quando o pipeline inteiro vai de Kafka para Kafka.

Serialização

O cliente traz serializers para strings, inteiros, longs, arrays de bytes e mais alguns tipos. Payloads reais usam JSON com schema, ou Avro ou Protocol Buffers por meio de um schema registry, cujos serializers colocam um identificador de schema na frente de cada registro para que os consumers possam evoluir de forma independente. Seja qual for o formato, falhas de desserialização no consumer são fatais para o loop de poll se não forem tratadas: o ErrorHandlingDeserializer do Spring Kafka, ou um wrapper que retorna um valor sentinela, permite que o loop pule um registro corrompido e siga em frente em vez de quebrar nele para sempre.

Spring Kafka

O Spring for Apache Kafka encapsula o mesmo cliente. KafkaTemplate.send retorna um CompletableFuture sobre o mesmo producer; @KafkaListener roda um loop de poll por thread do container, converte os registros e faz commit de acordo com um AckMode (BATCH depois de cada poll por padrão, MANUAL quando você mesmo chama Acknowledgment.acknowledge(), RECORD depois de 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" cria três threads consumidoras no grupo, cada uma com seu próprio KafkaConsumer, então esse valor não deve passar do número de partições. As propriedades spring.kafka.* do Boot configuram serializers, acks, o group id e o modo de acknowledgement do listener, e um DefaultErrorHandler com um DeadLetterPublishingRecoverer oferece o comportamento de “retry e depois dead-letter topic” que falta ao Kafka puro:

@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 um registro com falha bloqueia sua partição enquanto é reprocessado, mantenha o backoff curto e mande tudo o que exigir esperas longas para um topic de retry separado; o @RetryableTopic do Spring Kafka automatiza esse padrão.

Erros comuns

• acks=1 ou acks=0 “por desempenho”. Isso desabilita a idempotência e aceita perda quando o líder falha. Batching e compressão dão muito mais throughput sem essa perda.
• Fire-and-forget sem callback. send retorna antes de qualquer coisa acontecer; um registro com falha some se o callback ou o Future não for verificado.
• Processar em outra thread com auto-commit ligado. O poll seguinte commita registros que aquela thread ainda não terminou. Faça commit manual depois de processar, ou pause as partições.
• Consumers com efeito colateral não idempotente. At-least-once significa que duplicatas vão acontecer — depois de um rebalanceamento, de uma queda ou de um retry. Elimine duplicatas por uma chave ou por offset.
• Bloquear dentro do loop de poll por mais tempo que max.poll.interval.ms. O consumer é expulso, o grupo rebalanceia e o commit acaba falhando. Limite o trabalho por poll.
• Usar commitAsync em todo lugar. Seus commits não são repetidos, então uma falha transitória no desligamento faz perder o progresso. Termine com commitSync.
• Compartilhar um KafkaConsumer entre threads. O cliente detecta isso e lança ConcurrentModificationException; use um consumer por thread.

Principais pontos

• send adiciona a um batch; uma thread de I/O envia, repete as tentativas e confirma. Mantenha acks=all e a idempotência ligados, ajuste linger.ms, batch.size e a compressão para throughput e sempre verifique o callback.
• poll é o batimento cardíaco do consumer: continue chamando, limite o trabalho por chamada com max.poll.records e fique abaixo de max.poll.interval.ms.
• O offset commitado é o próximo registro a ser lido; faça commit depois de processar para ter at-least-once e torne o processamento idempotente, porque duplicatas depois de uma queda ou de um rebalanceamento não podem ser evitadas.
• Use commitAsync no loop para ganhar velocidade e commitSync na revogação e no desligamento; guarde os offsets no banco de destino quando precisar de exactly-once para um sistema que não é Kafka.
• Transações com transactional.id, sendOffsetsToTransaction e consumers em read_committed dão exactly-once para pipelines de Kafka para Kafka; o Kafka Streams entrega a mesma coisa pronta.
• KafkaTemplate, @KafkaListener, DefaultErrorHandler e o dead-letter recoverer do Spring Kafka encapsulam o cliente sem mudar nenhuma dessas regras.