Producteurs et consommateurs Kafka en Java : offsets, idempotence et transactions

Le client Java de Kafka cache une bonne dose de mécanique derrière deux classes. KafkaProducer.send ressemble à un simple appel de méthode, mais c'est la porte d'entrée d'un pipeline asynchrone de batching avec retries ; KafkaConsumer.poll a l'air de dire « donne-moi des messages », mais pilote aussi les heartbeats, l'appartenance au groupe et la gestion des offsets. Obtenir un comportement fiable suppose de comprendre ce que font ces deux appels et de choisir délibérément une poignée de réglages. Ce guide utilise org.apache.kafka:kafka-clients 3.x ; l'API est stable sur l'ensemble des versions 3.x et 4.x.

Le pipeline du producteur

Quand vous appelez send, l'enregistrement est sérialisé, affecté à une partition et ajouté à un batch en mémoire propre à cette partition. Un thread d'I/O en arrière-plan expédie les batches pleins — ou ceux plus anciens que linger.ms — vers les leaders des partitions, attend les acquittements selon acks, réessaie en cas d'échec, puis complète le Future et invoque votre callback. Au retour de send, rien n'a encore quitté le processus.

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
}

Les réglages qui comptent

L'idempotence est le réglage essentiel. Sans elle, un batch écrit par le leader mais dont l'acquittement s'est perdu est renvoyé et écrit deux fois, et un batch antérieur renvoyé peut atterrir derrière un batch plus récent. Avec elle, le broker reconnaît le renvoi à son numéro de séquence et conserve exactement une copie, dans l'ordre. Son coût n'est pas mesurable, et elle est activée par défaut avec acks=all — mais tout acks=1 ou acks=0 explicite la désactive silencieusement, ce qui explique souvent l'apparition de doublons après une montée de version.

send lève aussi une exception de manière synchrone dans quelques cas : échec de sérialisation, enregistrement trop gros pour max.request.size, et buffer plein au-delà de max.block.ms. Gérez à la fois l'exception et le callback.

La boucle de poll du consommateur

Un consommateur s'abonne à des topics, puis appelle poll dans une boucle. Chaque poll renvoie un lot d'enregistrements provenant des partitions attribuées à ce consommateur et, par effet de bord, envoie les heartbeats, rejoint ou réintègre le groupe et — avec l'auto-commit — commite les offsets. C'est à cause de cet effet de bord que la boucle doit continuer d'appeler poll même quand il n'y a rien à traiter.

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 ne s'applique que lorsque le groupe n'a aucun offset commité pour une partition : earliest traite l'historique, latest (la valeur par défaut) ne lit que les nouveaux enregistrements. Dès que le groupe a commité, le réglage n'a plus d'effet, ce qui surprend ceux qui le modifient sans rien voir changer.

max.poll.records borne la quantité de travail d'une itération, et max.poll.interval.ms la durée qu'elle peut prendre. Si traiter 200 enregistrements peut, dans le pire des cas, dépasser cinq minutes, baissez le premier ou relevez le second, ou bien déportez le travail sur un autre thread et appelez pause sur les partitions pendant son exécution, pour que poll puisse maintenir l'appartenance au groupe.

Arrêtez un consommateur depuis un autre thread avec consumer.wakeup(), qui fait lever une WakeupException au poll bloqué ; fermez ensuite le consommateur depuis son propre thread. La fermeture ne commite rien d'elle-même, sauf si l'auto-commit est actif : le chemin d'arrêt doit donc commiter explicitement.

Commiter les offsets

L'offset que vous commitez est la position du prochain enregistrement à lire. L'endroit de la boucle où vous commitez détermine la taille de la fenêtre de doublons ou de perte après un crash.

Un schéma courant combine les deux formes : commitAsync dans la boucle pour le débit, et commitSync dans le bloc finally et dans onPartitionsRevoked pour la correction. Quel que soit votre choix, le consommateur doit tolérer de voir un enregistrement deux fois — un crash entre le traitement et le commit est impossible à empêcher — et le traitement doit donc être idempotent, en général en enregistrant (topic, partition, offset) ou une clé métier à côté du résultat.

Si les offsets ont plutôt leur place avec les données — dans la même transaction de base de données que la ligne produite par l'enregistrement —, désactivez l'auto-commit, écrivez l'offset en base et, au démarrage, relisez-le et faites un seek de chaque partition attribuée vers cet offset depuis onPartitionsAssigned. Vous obtenez ainsi de l'exactly-once entre Kafka et cette base, sans transactions Kafka.

Transactions et exactly-once

Pour les pipelines qui lisent dans Kafka, traitent, puis réécrivent dans Kafka, les transactions rendent atomiques les enregistrements produits et le commit des offsets d'entrée : soit les sorties sont visibles et les offsets commités, soit ni l'un ni l'autre. Combinées à l'idempotence, elles forment la sémantique 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 identifie le producteur d'un redémarrage à l'autre : une nouvelle instance avec le même identifiant isole (fences) l'ancienne, si bien qu'une instance zombie ne peut pas commiter du travail périmé. Les consommateurs en aval ne voient les sorties qu'à condition d'être en read_committed ; avec la valeur par défaut read_uncommitted, ils verraient aussi les enregistrements des transactions annulées. Les transactions ont un coût en latence — un commit implique un aller-retour avec le coordinateur de transactions plus des marqueurs dans chaque partition écrite — et conviennent mieux à des lots d'enregistrements qu'à des enregistrements isolés.

Kafka Streams active tout cela avec un seul réglage (processing.guarantee=exactly_once_v2), ce qui est généralement la voie la plus simple quand tout le pipeline va de Kafka à Kafka.

Sérialisation

Le client fournit des sérialiseurs pour les chaînes, les entiers, les longs, les tableaux d'octets et quelques autres types. Les vrais payloads utilisent du JSON avec un schéma, ou Avro ou Protocol Buffers via un schema registry, dont les sérialiseurs préfixent chaque enregistrement par un identifiant de schéma afin que les consommateurs puissent évoluer indépendamment. Quel que soit le format, un échec de désérialisation côté consommateur est fatal à la boucle de poll s'il n'est pas géré : l'ErrorHandlingDeserializer de Spring Kafka, ou un wrapper qui renvoie une valeur sentinelle, permet à la boucle d'ignorer un enregistrement corrompu et de continuer au lieu de planter dessus indéfiniment.

Spring Kafka

Spring for Apache Kafka enveloppe le même client. KafkaTemplate.send renvoie un CompletableFuture au-dessus du même producteur ; @KafkaListener exécute une boucle de poll par thread de conteneur, convertit les enregistrements et commite selon un AckMode (BATCH après chaque poll par défaut, MANUAL quand vous appelez vous-même Acknowledgment.acknowledge(), RECORD après chaque enregistrement).

@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" crée trois threads consommateurs dans le groupe, chacun avec son propre KafkaConsumer ; cette valeur ne doit donc pas dépasser le nombre de partitions. Les propriétés spring.kafka.* de Boot configurent les sérialiseurs, acks, le group id et le mode d'acquittement du listener, et un DefaultErrorHandler associé à un DeadLetterPublishingRecoverer apporte le comportement « retry puis dead-letter topic » qui manque à Kafka brut :

@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
}

Comme un enregistrement en échec bloque sa partition pendant qu'il est relancé, gardez un backoff court et déplacez tout ce qui nécessite de longues attentes vers un topic de retry distinct ; le @RetryableTopic de Spring Kafka automatise ce schéma.

Erreurs courantes

• acks=1 ou acks=0 « pour les performances ». Cela désactive l'idempotence et accepte la perte de données en cas de défaillance du leader. Le batching et la compression apportent bien plus de débit, sans perte.
• Fire-and-forget sans callback. send rend la main avant que quoi que ce soit ne se produise ; un enregistrement en échec disparaît si le callback ou le Future n'est pas vérifié.
• Traiter dans un autre thread avec l'auto-commit actif. Le poll suivant commite des enregistrements que ce thread n'a pas terminés. Commitez manuellement après traitement, ou mettez les partitions en pause.
• Des consommateurs sans effet de bord idempotent. At-least-once signifie que les doublons arriveront — après un rééquilibrage, un crash ou un retry. Dédoublonnez sur une clé ou un offset.
• Bloquer dans la boucle de poll plus longtemps que max.poll.interval.ms. Le consommateur est évincé, le groupe se rééquilibre et le commit finit par échouer. Bornez le travail par poll.
• Utiliser commitAsync partout. Ses commits ne sont pas relancés, donc une erreur transitoire à l'arrêt fait perdre la progression. Terminez par un commitSync.
• Partager un KafkaConsumer entre plusieurs threads. Le client le détecte et lève une ConcurrentModificationException ; utilisez un consommateur par thread.

Points clés à retenir

• send ajoute à un batch ; un thread d'I/O expédie, relance et acquitte. Gardez acks=all et l'idempotence, ajustez linger.ms, batch.size et la compression pour le débit, et vérifiez toujours le callback.
• poll est le battement de cœur du consommateur : continuez à l'appeler, bornez le travail par appel avec max.poll.records et restez sous max.poll.interval.ms.
• L'offset commité est le prochain enregistrement à lire ; commitez après traitement pour de l'at-least-once et rendez le traitement idempotent, car les doublons après un crash ou un rééquilibrage sont inévitables.
• Utilisez commitAsync pour la vitesse dans la boucle et commitSync lors de la révocation et de l'arrêt ; stockez les offsets dans la base cible quand vous avez besoin d'exactly-once vers un système autre que Kafka.
• Les transactions avec transactional.id, sendOffsetsToTransaction et des consommateurs en read_committed apportent l'exactly-once aux pipelines de Kafka à Kafka ; Kafka Streams offre la même chose clés en main.
• KafkaTemplate, @KafkaListener, DefaultErrorHandler et le dead-letter recoverer de Spring Kafka enveloppent le client sans changer aucune de ces règles.