Kafka producer : acks, idempotence et batching

1. Aperçu

Envoyer un message vers Kafka ne nécessite qu’une ligne send(). La partie difficile réside dans les questions qui s’ensuivent : le message a-t-il vraiment été sauvegardé, sera-t-il écrit en doublon lors d’une nouvelle tentative du producteur, et comment envoyer des dizaines de milliers de messages par seconde sans surcharger le broker ?

Cet article explore trois groupes de configuration qui répondent à ces trois questions : acks, enable.idempotence et le batching (batch.size, linger.ms, compression.type). Une fois cet article lu, vous saurez comment configurer un producteur à la fois sûr et rapide, avec le client Kafka pur et Spring Boot 3.x, et comment le tester.

2. Configuration initiale

Avec Java pur, il vous suffit de kafka-clients :

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

Avec Spring Boot 3.x, utilisez spring-kafka. La version est gérée par Spring Boot, et elle entraîne déjà kafka-clients :

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

Tout au long de cet article, nous enverrons des événements de commande :

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 : quand un message est-il considéré comme envoyé ?

acks définit le nombre de confirmations que le producteur doit attendre du broker avant de considérer une requête comme réussie.

Valeur	Signification	Risque
acks=0	Ne pas attendre	Perte de message sans le savoir
acks=1	Attendre la fin de l’écriture du leader	Perte de message si le leader meurt avant la réplication
acks=all (-1)	Attendre l’écriture de tous les réplicas dans l’ISR	Latence légèrement plus élevée

Exemple de configuration avec le client pur :

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

L’appel de .get() rend l’envoi synchrone. Ici, nous le faisons seulement pour imprimer le résultat. Avec acks=all, get() ne revient que lorsque tous les réplicas de l’ISR ont terminé l’écriture :

partition=2 offset=17

Nous utilisons customerId comme clé, donc toutes les commandes du même client vont dans la même partition et conservent l’ordre correct.

3.1. acks=all n’est pas suffisant sans min.insync.replicas

acks=all signifie « tous les réplicas actuellement dans l’ISR ». Supposons que l’ISR se réduise à seulement le leader. Dans ce cas, acks=all ne diffère pas de acks=1. Vous devez configurer également sur le topic ou le broker :

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

Avec replication-factor=3 et min.insync.replicas=2, chaque écriture doit avoir au moins 2 réplicas. Si l’ISR tombe en dessous de 2, le producteur reçoit une NotEnoughReplicasException au lieu d’écrire silencieusement une seule copie.

Note : Ne fixez pas min.insync.replicas égal à replication-factor. Dans ce cas, il suffit d’un simple redémarrage du broker pour que le topic refuse toutes les écritures.

4. Idempotence : retry sans duplication d’écriture

acks=all entraîne des retries. Considérons le scénario suivant : le broker a écrit le message mais la réponse est perdue en chemin. Le producteur pense que l’envoi a échoué, alors il le renvoie, et le message est écrit deux fois.

Le producteur idempotent résout ce scénario exact. Lorsque vous activez enable.idempotence=true :

• Le broker attribue un ID producteur (PID) au producteur.
• Chaque lot envoyé vers une partition porte un numéro de séquence croissant.
• Le broker supprime les lots avec un numéro de séquence déjà vu et rejette les lots qui sautent dans la séquence. Ainsi, l’ordre est préservé même lorsque plusieurs requêtes sont en vol.

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

La signification de chaque ligne :

• ENABLE_IDEMPOTENCE_CONFIG active le mécanisme PID et numéro de séquence.
• L’idempotence force acks=all. Si vous définissez explicitement acks=1 avec enable.idempotence=true, le producteur lève une ConfigException à l’initialisation.
• max.in.flight.requests.per.connection doit être ≤ 5. Dans cette limite, Kafka garantit toujours l’ordre dans la partition.
• Mettez retries très élevé. Le délai d’attente réel des retries est limité par delivery.timeout.ms (par défaut 2 minutes).

À partir de Kafka 3.0, acks=all et enable.idempotence=true sont par défaut. Néanmoins, vous devez déclarer explicitement. Sinon, il suffit que quelqu’un mette acks=1 pour que l’idempotence soit silencieusement désactivée.

4.1. Ce que l’idempotence ne peut pas faire

Voici la partie souvent mal comprise :

• Il ne prévient les doublons que dans une session de producteur, par partition. Si l’application redémarre, le producteur reçoit un nouveau PID. Les messages renvoyés après un redémarrage sont toujours écrits deux fois.
• Il ne prévient pas les doublons causés par votre code appelant send() deux fois, par exemple si vous effectuez vous-même un retry dans catch.
• Il n’offre pas d’écriture atomique sur plusieurs partitions.

Ces cas nécessitent des transactions (transactional.id) ou un consumer capable de dédupliquer par orderId.

5. Batching : échanger la latence pour le débit

Le producteur n’envoie pas message par message. send() place simplement l’enregistrement dans un buffer en mémoire, organisé par partition. Un thread de fond (sender) groupe les enregistrements en lots puis les envoie. Trois paramètres principaux :

Paramètre	Par défaut	Rôle
batch.size	16384 byte	Taille maximale d’un lot pour une partition
linger.ms	0 (5 à partir de Kafka 4.0)	Temps d’attente supplémentaire pour remplir le lot
compression.type	none	Compresser le lot entier : gzip, snappy, lz4, zstd
buffer.memory	32 MB	Mémoire totale pour tous les lots en attente

Un lot est envoyé quand il est plein (batch.size) ou quand linger.ms expire, selon ce qui se produit en premier. Avec linger.ms=0, le lot est presque toujours envoyé immédiatement, il ne contient donc généralement que quelques enregistrements.

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

Cette configuration accepte une latence supplémentaire d’environ 10 ms au maximum. En échange, chaque requête porte 64 KB de données compressées. Avec des données JSON répétitives comme les événements de commande, lz4 ou zstd réduit généralement considérablement la bande passante réseau et l’espace disque sur le broker. La compression par lot est beaucoup plus efficace que la compression par message, car plus le lot est grand, plus il y a de chaînes répétitives à exploiter.

Vous pouvez vérifier cela par les métriques du producteur :

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

En envoyant 10 000 commandes consécutives avec linger.ms=0, puis en répétant avec linger.ms=10, vous verrez records-per-request-avg augmenter sensiblement :

# 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

Les chiffres exacts dépendent de la machine et de la charge, mais la tendance est toujours la même.

Note : Si le buffer est plein (la lenteur du consumer n’est pas pertinente ici, il s’agit du broker lent ou du réseau congestionné), send() va bloquer jusqu’à max.block.ms (60 secondes par défaut) puis lancer une TimeoutException. N’appelez pas send() dans un thread traitant une requête HTTP sans tenir compte de cette possibilité.

Note : delivery.timeout.ms doit être ≥ linger.ms + request.timeout.ms. Sinon, le producteur refuse de s’initialiser.

6. Configuration dans Spring Boot 3.x

Spring Boot a des propriétés dédiées pour les paramètres courants. Les autres paramètres passent par 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

Le publisher utilise KafkaTemplate. À partir de Spring Kafka 3.0, send() retourne un 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("L'envoi de la commande {} a échoué", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Commande {} -> partition {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

Notez trois points :

• Le callback s’exécute sur le thread sender du producteur. Ne faites pas de travail lourd ni n’appelez de blocage dans whenComplete, car cela ralentirait tous les autres envois.
• Quand le callback reçoit une erreur, le producteur a épuisé toutes ses tentatives dans la limite de delivery.timeout.ms. Ne relancez pas manuellement en rappelant send() immédiatement. C’est précisément la source de doublons que l’idempotence ne peut pas bloquer.
• Retourner un CompletableFuture permet à l’appelant de décider s’il faut attendre ou non.

7. Test du comportement du producteur

kafka-clients fournit MockProducer pour tester sans broker. Pour l’utiliser, nous séparons la logique dans une classe qui accepte l’interface 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;
    }
}

Test avec JUnit 5 et 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();
    }
}

Chaque test vérifie un comportement :

• Le premier test utilise autoComplete=true, donc chaque send() réussit immédiatement. history() permet de vérifier la clé, confirmant que l’ordre par client est préservé.
• Le deuxième test utilise autoComplete=false puis errorNext(...) pour simuler un broker refusant l’écriture. C’est comment tester votre code de gestion d’erreur.
• Le dernier test initialise ProducerConfig pour que Kafka valide lui-même la combinaison de configuration. Une combinaison incorrecte, comme acks=1 avec l’idempotence, fera échouer le test avec une ConfigException au lieu d’exploser au déploiement. Kafka normalise all en "-1", donc l’assertion compare avec "-1".

8. Quand faut-il assouplir ?

La configuration “maximalement sûre” n’est pas toujours correcte :

• Métriques, logs, clickstream : la perte de quelques enregistrements est acceptable. Dans ce cas, acks=1, linger.ms élevé et compression zstd pour le meilleur débit.
• Systèmes nécessitant une faible latence par message (par exemple, cotations en temps réel) : gardez linger.ms petit (0–2 ms). Dans ce cas, n’attendez pas de grands lots.
• Paiements, commandes : utilisez toujours acks=all, idempotence et min.insync.replicas=2. Si vous avez besoin de lire-traiter-écrire exactement une fois, envisagez les transactions.

Note : Définir max.in.flight.requests.per.connection=1 « pour être sûr » n’est pas nécessaire lorsque l’idempotence est activée. Cela réduit simplement le débit.

9. Conclusion

Trois groupes de configuration résolvent trois problèmes différents :

• acks=all associé à min.insync.replicas garantit qu’aucun message n’est perdu.
• enable.idempotence garantit que les retries ne causent pas de doublons et préservent l’ordre dans la partition. Son champ d’application est limité à une session de producteur.
• batch.size, linger.ms et compression.type déterminent le débit, en échange d’une légère latence.

Déclarez explicitement ces valeurs au lieu de vous fier aux valeurs par défaut. Validez-les par des tests pour éviter les surprises au déploiement. Et ne relancez pas vous-même send() dans le code quand le producteur le fait déjà pour vous.