Envoyer et recevoir des messages Kafka dans Spring Boot

1. Vue d’ensemble

Apache Kafka est une plateforme de streaming d’événements populaire pour permettre aux services de communiquer de façon asynchrone. Spring Boot prend en charge Kafka via la bibliothèque Spring Kafka, ce qui vous permet de ne pas avoir à gérer manuellement KafkaProducer et la boucle poll().

Dans cet article, vous allez construire un petit exemple : un service de commande envoie un événement OrderCreatedEvent, et un consumer reçoit cet événement pour réserver des articles en stock. À la fin, vous saurez comment configurer, envoyer et recevoir des messages au format JSON, gérer les erreurs avec Dead Letter Topic et tester avec Embedded Kafka.

2. Préparation

2.1. Dépendances Maven

Vous avez besoin de spring-kafka. Spring Boot gère déjà la version, vous n’avez donc pas besoin de la spécifier :

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

<!-- Pour les tests -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test fournit un broker Kafka en mémoire pour les tests. Awaitility aide à attendre le traitement asynchrone sans avoir à utiliser Thread.sleep().

2.2. Lancer Kafka localement

Le plus rapide est d’utiliser l’image officielle d’Apache Kafka en mode KRaft, c’est-à-dire sans ZooKeeper :

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

Le broker écoutera sur localhost:9092.

3. Modèle de données et configuration

L’événement envoyé est un simple record :

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

Ensuite voici la configuration dans application.yml :

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

Signification de chaque partie :

• bootstrap-servers est l’adresse du broker.
• Le producer utilise StringSerializer pour la clé et JsonSerializer pour la valeur. JsonSerializer utilise Jackson pour convertir l’objet en JSON.
• acks: all force le broker à confirmer uniquement après que toutes les répliques synchronisées aient écrit. C’est plus sûr mais un peu plus lent.
• group-id identifie le consumer group. Les instances du même groupe partageront les partitions.
• auto-offset-reset: earliest fait que le consumer lira depuis le début du topic quand le groupe n’a pas encore d’offset commit.
• spring.json.trusted.packages déclare les packages que JsonDeserializer est autorisé à instancier. Sans cette ligne, vous rencontrerez l’erreur “The class … is not in the trusted packages”.

3.1. Créer des topics automatiquement

Spring Boot configure automatiquement un KafkaAdmin. Ce bean créera tout NewTopic déclaré dans le contexte :

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

Au démarrage de l’application, le topic orders avec 3 partitions sera créé s’il n’existe pas. Si le topic existe déjà, KafkaAdmin l’ignorera et ne remplacera pas la configuration existante.

4. Envoyer des messages avec KafkaTemplate

Spring Boot crée automatiquement un bean KafkaTemplate basé sur la configuration du producer. Vous avez juste besoin de l’injecter :

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Impossible d'envoyer la commande {}", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("Commande {} envoyée à la partition {} offset {}",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

Points à noter :

• send(topic, key, value) utilise orderId comme clé. Kafka hache la clé pour choisir une partition, donc tous les messages d’une même commande iront dans la même partition. Cela garantit que leur ordre est préservé.
• Depuis Spring Kafka 3.0, send() retourne CompletableFuture<SendResult> au lieu de ListenableFuture qui a été supprimé.
• send() est asynchrone. L’appel retourne immédiatement, tandis que le callback dans whenComplete s’exécute quand le broker répond.

Quand l’envoi réussit, le log ressemble à :

INFO  c.n.s.order.OrderEventPublisher : Commande ORD-1001 envoyée à la partition 2 offset 0

Si vous avez besoin de bloquer le thread jusqu’à ce que l’envoi soit terminé, par exemple dans une tâche batch, vous pouvez appeler publish(event).get(5, TimeUnit.SECONDS). Ne faites pas cela dans un thread de traitement de requête à charge élevée, car chaque appel gardera un thread en attente du broker.

Note : Écrire en base de données puis envoyer Kafka dans la même méthode n’est pas une transaction atomique. Si la base de données est commit avec succès mais que l’envoi Kafka échoue, les données seront incohérentes. Pour les données importantes, vous devriez envisager le pattern Transactional Outbox.

5. Recevoir des messages avec @KafkaListener

Le côté consumer n’a besoin que d’une méthode avec l’annotation @KafkaListener :

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("Commande {} reçue de la partition {} offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService est juste un service ordinaire :

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("Montant invalide : " + event.amount());
        }
        // Réserver le stock...
    }
}

Explication :

• topics spécifie le topic à écouter. Le groupId provient de spring.kafka.consumer.group-id, ou vous pouvez le redéfinir directement dans l’annotation.
• concurrency = "3" crée 3 threads consumer, chacun gérant une partition. Une valeur supérieure au nombre de partitions est inutile, car les threads excédentaires ne recevront rien.
• @Payload reçoit l’objet déjà désérializé. @Header récupère les métadonnées comme la partition et l’offset.
• Par défaut, l’offset est commit après que la méthode s’exécute sans lever d’exception (AckMode.BATCH). Donc si l’application meurt au milieu, le message sera traité à nouveau. C’est la sémantique at-least-once, donc votre logique doit être idempotente : traiter deux fois le même orderId ne doit pas réserver deux fois.

Si vous avez besoin de la clé, du timestamp ou d’autres headers personnalisés, vous pouvez recevoir l’entier ConsumerRecord<String, OrderCreatedEvent> comme paramètre.

6. Gestion des erreurs et Dead Letter Topic

Quand le listener lève une exception, le DefaultErrorHandler par défaut réessayera 9 fois de suite, puis enregistrera un log et ignorera le message. En pratique, vous voudrez souvent déplacer les messages en erreur vers un topic dédié pour inspection ultérieure :

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Boot attache automatiquement ce bean CommonErrorHandler à la container factory par défaut. La configuration fonctionne comme suit :

• FixedBackOff(1000L, 2) réessaye au maximum 2 fois, avec 1 seconde entre chaque tentative.
• Quand les tentatives s’épuisent, DeadLetterPublishingRecoverer envoie le message au topic orders.DLT. Ce nom est le nom d’origine plus le suffixe .DLT, et le message conserve sa partition. Le topic DLT a donc besoin d’au moins autant de partitions que le topic d’origine.
• IllegalArgumentException est une erreur de données donc réessayer n’aide pas. Le handler enverra directement le message au DLT.

Un message avec amount = -5 produira un log similaire à :

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : Message d'erreur dans DLT : ORD-1002

Note : Les erreurs qui surviennent pendant la désérialisation, par exemple du JSON malformé, ne pourront pas atteindre le listener. Si vous ne les gérez pas, le consumer peut rester bloqué sur ce message indéfiniment, ce qu’on appelle une “poison pill”. Enveloppez le deserializer avec ErrorHandlingDeserializer :

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

Avec cette configuration, les erreurs de désérialisation sont passées à DefaultErrorHandler et le message en erreur va aussi au DLT.

7. Test avec Embedded Kafka

@EmbeddedKafka démarre un broker en mémoire, vous n’avez donc pas besoin de Docker pour les tests. L’exemple ci-dessous teste la boucle complète : envoyer via OrderEventPublisher, puis vérifier que InventoryService a reçu l’événement correctement.

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

Chaque partie fait quoi :

• ${spring.embedded.kafka.brokers} est une propriété fournie par @EmbeddedKafka. Elle contient l’adresse du broker embarqué, et on l’utilise pour redéfinir bootstrap-servers.
• @MockitoBean (disponible depuis Spring Boot 3.4) remplace l’InventoryService réel par un mock pour faciliter la vérification. Avec des versions plus anciennes, utilisez @MockBean.
• publish(event).get() attend la confirmation du broker, donc le test s’assure que le message a été écrit.
• Le listener s’exécute sur un autre thread, donc await().untilAsserted(...) réessayera la vérification pendant au maximum 10 secondes.
• Puisque OrderCreatedEvent est un record, equals() compare par valeur. assertThat(...).isEqualTo(event) vérifie donc que le JSON a été correctement sérialisé et désérialisé.

C’est auto-offset-reset: earliest dans la configuration qui rend le test stable. Si le consumer rejoint le groupe après que le message a été envoyé, il peut quand même lire ce message.

8. Erreurs courantes et quand ne pas utiliser Kafka

Quelques erreurs que vous risquez de rencontrer :

• trusted.packages manquant : JsonDeserializer refuse de créer une classe inconnue. Déclarez le bon package. N’utilisez "*" que si vous contrôlez entièrement tous les producers.
• Producer et consumer dans des packages différents : JsonSerializer écrit par défaut le nom de classe complet dans le header __TypeId__. Si le service consumer a une classe du même nom mais dans un package différent, la désérialisation échouera. Dans ce cas, définissez spring.json.use.type.headers: false et spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent du côté consumer.
• Le listener prend longtemps : si un poll() prend plus de max.poll.interval.ms (par défaut 5 minutes), le broker considère que le consumer est mort et déclenche un rebalance. Le résultat est que les messages sont traités deux fois. Réduisez max.poll.records ou optimisez votre logique.
• concurrency supérieur au nombre de partitions : les threads en excès ne feront rien. Pour augmenter le débit, vous devez d’abord augmenter le nombre de partitions.
• Envoyer une clé null : le message est distribué uniformément entre les partitions, vous perdez l’ordre par entité.

Kafka n’est pas non plus le bon choix dans tous les cas :

• Vous avez besoin d’une communication synchrone request/response avec une réponse immédiate. Utilisez plutôt REST ou gRPC.
• Votre système est petit, le débit est faible et vous n’avez pas besoin de conserver l’historique des événements. Exploiter un cluster Kafka est alors un surcoût, et RabbitMQ ou une simple table queue en base de données peut suffire.
• Vous avez besoin d’un routage complexe selon chaque message, ou avez besoin de latence ou de priorité différentes par message. C’est la force des message brokers traditionnels.

9. Conclusion

Dans cet article, vous avez configuré Spring Kafka avec un sérialiseur JSON et créé des topics avec NewTopic. Vous avez envoyé des messages via KafkaTemplate avec une clé pour préserver l’ordre, et reçu des messages avec @KafkaListener ayant un concurrency. Vous avez aussi géré les erreurs avec DefaultErrorHandler combiné à Dead Letter Topic, puis écrit un test d’intégration avec @EmbeddedKafka et Awaitility.

Quand vous déployez en production, n’oubliez pas trois choses : le listener doit être idempotent, vous devez gérer les erreurs de désérialisation avec ErrorHandlingDeserializer, et l’envoi d’événements doit être cohérent avec les données en base de données.