Event sourcing avec Kafka : enregistrer les événements plutôt que l'état

1. Vue d’ensemble

La plupart des applications n’enregistrent que l’état actuel. Quand une commande est annulée, la ligne de la table orders devient CANCELLED. Personne ne se souvient plus de ce qu’il y avait avant et si le paiement était fait.

Event sourcing fait le contraire : on enregistre seulement la séquence d’événements qui se sont produits, l’état est le résultat du calcul effectué sur cette séquence. Dans cet article, vous allez utiliser Kafka comme event log pour l’aggregate Order avec Spring Boot 3.x. Vous allez reconstruire l’état en rejouant les événements, construire un read model par projection et maîtriser les limites pratiques de cette approche.

2. Enregistrer l’état versus enregistrer les événements

Avec l’approche traditionnelle, une commande est un enregistrement :

orders: id=A1 | status=CANCELLED | total=250000

Avec event sourcing, cette même commande est une séquence d’événements uniquement en ajout, jamais modifiée ou supprimée :

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="Client a changé d'avis")

L’état CANCELLED, total=250000 est calculé en « pliant » (fold) les événements dans l’ordre. Cette approche d’enregistrement offre trois avantages :

Audit naturel : l’historique complet est la donnée source elle-même.
Nouveau read model sans migration : pour ajouter un rapport, vous rejouez le log dans une nouvelle projection.
Débogage dans le temps : vous pouvez reconstruire l’état à n’importe quel moment.

Kafka est bien adapté à ce rôle de log. Un topic est une séquence append-only ordonnée par partition. Si vous utilisez orderId comme clé, tous les événements de la même commande tombent dans la même partition et conservent le bon ordre.

3. Configuration du projet

3.1. Dépendances

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Spring Boot gère la version de spring-kafka, donc vous n’avez pas besoin de déclarer <version>. spring-boot-starter-json apporte Jackson et le module jsr310 pour sérialiser Instant.

3.2. Configuration

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
      properties:
        enable.idempotence: true
        spring.json.add.type.headers: false
    consumer:
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.value.default.type: com.nonla.orders.OrderEvent
        spring.json.trusted.packages: com.nonla.orders
        spring.json.use.type.headers: false

Points notables :

acks: all et enable.idempotence: true garantissent que les événements sont écrits durablement et que les relances du producer ne créent pas de doublons.
On désactive l’en-tête de type de Spring. Le type d’événement se trouve directement dans le JSON via la propriété type, donc un consumer écrit dans un autre langage peut toujours le lire.
auto-offset-reset: earliest permet à un nouveau consumer group de relire depuis le début du log. C’est une condition nécessaire pour le replay.

3.3. Topic gardant les événements à titre permanent

@Configuration
public class KafkaTopicConfig {

    @Bean
    NewTopic orderEvents() {
        return TopicBuilder.name("order-events")
                .partitions(3)
                .replicas(1)
                .config(TopicConfig.RETENTION_MS_CONFIG, "-1")
                .config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
                .build();
    }
}

Par défaut, Kafka supprime les données après 7 jours. Avec un event store, cela signifie perdre la donnée source. retention.ms=-1 désactive la suppression basée sur le temps.

Remarque : N’utilisez pas cleanup.policy=compact pour le topic d’événements. Le compaction ne garde que le dernier enregistrement pour chaque clé, c’est-à-dire supprime tout l’historique de la commande. Le compaction convient seulement pour un topic snapshot.

4. Définir les événements

Un événement est une donnée immuable qui décrit quelque chose qui s’est produit et est nommé au passé. sealed interface combiné avec record de Java 21 est très approprié pour cela :

package com.nonla.orders;

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
        @JsonSubTypes.Type(value = OrderEvent.Created.class, name = "OrderCreated"),
        @JsonSubTypes.Type(value = OrderEvent.ItemAdded.class, name = "ItemAdded"),
        @JsonSubTypes.Type(value = OrderEvent.Paid.class, name = "OrderPaid"),
        @JsonSubTypes.Type(value = OrderEvent.Cancelled.class, name = "OrderCancelled")
})
public sealed interface OrderEvent {

    String orderId();
    Instant occurredAt();

    record Created(String orderId, String customerId, Instant occurredAt) implements OrderEvent {}
    record ItemAdded(String orderId, String sku, BigDecimal price, Instant occurredAt) implements OrderEvent {}
    record Paid(String orderId, BigDecimal amount, Instant occurredAt) implements OrderEvent {}
    record Cancelled(String orderId, String reason, Instant occurredAt) implements OrderEvent {}
}

Quand vous envoyez un ItemAdded, le payload sur Kafka ressemble à :

{"type":"ItemAdded","orderId":"A1","sku":"COFFEE","price":150000,"occurredAt":"2026-10-11T03:15:00Z"}

Le nom dans @JsonSubTypes est un contrat à long terme. Vous pouvez changer le nom de la classe Java, mais pas la chaîne "ItemAdded", sinon les anciens événements ne pourront pas être relus.

5. Aggregate : reconstruire l’état à partir des événements

Order a deux types de méthodes :

Command (addItem, pay, cancel) : vérifient les règles métier puis retournent un événement. Les commands ne modifient pas elles-mêmes l’état.
apply : reçoit un événement et met à jour l’état. Cette méthode ne doit pas lever d’erreur métier, car l’événement s’est déjà produit.

public final class Order {

    public enum Status { NEW, PAID, CANCELLED }

    private String id;
    private String customerId;
    private Status status;
    private BigDecimal total = BigDecimal.ZERO;
    private final List<String> items = new ArrayList<>();
    private long version;

    public static Order replay(List<OrderEvent> history) {
        Order order = new Order();
        history.forEach(order::apply);
        return order;
    }

    public void apply(OrderEvent event) {
        switch (event) {
            case OrderEvent.Created e -> { id = e.orderId(); customerId = e.customerId(); status = Status.NEW; }
            case OrderEvent.ItemAdded e -> { items.add(e.sku()); total = total.add(e.price()); }
            case OrderEvent.Paid e -> status = Status.PAID;
            case OrderEvent.Cancelled e -> status = Status.CANCELLED;
        }
        version++;
    }

    public OrderEvent addItem(String sku, BigDecimal price) {
        requireStatus(Status.NEW, "Peut uniquement ajouter des articles à une nouvelle commande");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "La commande n'est pas à l'état en attente de paiement");
        if (total.signum() == 0) {
            throw new IllegalStateException("Impossible de payer une commande vide");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("La commande a déjà été annulée");
        }
        return new OrderEvent.Cancelled(id, reason, Instant.now());
    }

    private void requireStatus(Status expected, String message) {
        if (status != expected) throw new IllegalStateException(message);
    }

    public String id() { return id; }
    public Status status() { return status; }
    public BigDecimal total() { return total; }
    public List<String> items() { return List.copyOf(items); }
    public long version() { return version; }
}

Puisque OrderEvent est sealed, switch dans apply est entièrement vérifié par le compilateur. Quand vous ajoutez un nouveau type d’événement mais oubliez de le traiter, le code ne compile pas. C’est un filet de sécurité précieux.

5.1. Tester l’aggregate

L’aggregate ne dépend pas de Kafka, donc les tests sont très rapides :

class OrderTest {

    private final Instant now = Instant.parse("2026-10-11T03:00:00Z");

    @Test
    void replayRebuildsStateFromHistory() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.ItemAdded("A1", "COFFEE", new BigDecimal("150000"), now),
                new OrderEvent.ItemAdded("A1", "MUG", new BigDecimal("100000"), now),
                new OrderEvent.Cancelled("A1", "Client a changé d'avis", now)));

        assertThat(order.status()).isEqualTo(Order.Status.CANCELLED);
        assertThat(order.total()).isEqualByComparingTo("250000");
        assertThat(order.items()).containsExactly("COFFEE", "MUG");
        assertThat(order.version()).isEqualTo(4);
    }

    @Test
    void cannotPayCancelledOrder() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.Cancelled("A1", "Rupture de stock", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("attendre le paiement");
    }
}

Le premier test montre que quatre événements sont pliés en un état cohérent, avec version égal au nombre d’événements appliqués. Le deuxième test montre que les règles métier sont vérifiées sur la base de l’état reconstruit à partir de l’historique.

6. Écrire les événements dans Kafka

6.1. Event store

Pour traiter une command, vous avez besoin de l’historique d’une commande. Kafka ne supporte pas les requêtes « donne-moi tous les enregistrements avec la clé A1 », donc on garde une copie en mémoire. Cette copie est chargée en relisant tout le topic à chaque démarrage de l’application :

@Component
public class OrderEventStore {

    private static final String TOPIC = "order-events";

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final Map<String, List<OrderEvent>> streams = new ConcurrentHashMap<>();

    public OrderEventStore(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void append(OrderEvent event) {
        try {
            kafkaTemplate.send(TOPIC, event.orderId(), event).get(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new IllegalStateException("Impossible d'écrire l'événement " + event, e);
        }
    }

    public List<OrderEvent> load(String orderId) {
        return List.copyOf(streams.getOrDefault(orderId, List.of()));
    }

    @KafkaListener(topics = TOPIC, groupId = "order-store-#{T(java.util.UUID).randomUUID()}")
    void on(OrderEvent event) {
        streams.computeIfAbsent(event.orderId(), id -> new CopyOnWriteArrayList<>()).add(event);
    }
}

Explication de chaque partie :

send(TOPIC, event.orderId(), event) utilise orderId comme clé. Ainsi, tous les événements d’une commande se trouvent dans la même partition et conservent le bon ordre.
.get(...) attend la confirmation du broker. Une command n’est considérée comme réussie que quand l’événement est dans le log.
Le listener utilise un groupId aléatoire. À chaque démarrage, c’est un nouveau group, combiné avec earliest, donc tout l’historique est relu dans la map.

6.2. Service Command

@Service
public class OrderCommandService {

    private final OrderEventStore store;

    public OrderCommandService(OrderEventStore store) {
        this.store = store;
    }

    public String createOrder(String customerId) {
        String orderId = UUID.randomUUID().toString();
        store.append(new OrderEvent.Created(orderId, customerId, Instant.now()));
        return orderId;
    }

    public void addItem(String orderId, String sku, BigDecimal price) {
        Order order = loadExisting(orderId);
        store.append(order.addItem(sku, price));
    }

    public void pay(String orderId) {
        store.append(loadExisting(orderId).pay());
    }

    private Order loadExisting(String orderId) {
        List<OrderEvent> history = store.load(orderId);
        if (history.isEmpty()) {
            throw new NoSuchElementException("Commande introuvable " + orderId);
        }
        return Order.replay(history);
    }
}

Le flux de traitement est toujours le même : charger l’historique → rejouer → appeler la command → ajouter l’événement. Pas d’instruction UPDATE du tout.

Remarque : La map en mémoire est mise à jour de manière asynchrone via le listener. Si vous appelez addItem juste après createOrder, la map n’a peut-être pas reçu Created. De plus, Kafka n’a pas d’optimistic concurrency du style « écrire seulement si la version actuelle est 4 ». Deux instances traitant la même commande peuvent écrire deux événements contradictoires. En production, vous devriez acheminer tous les commands d’un orderId vers un seul consumer (le command topic a une clé selon orderId), ou utiliser Kafka Streams avec un state store. Une autre option est d’utiliser un event store dédié ou une base de données comme source d’écriture principale, puis de publier sur Kafka.

7. Projection : construire le read model

L’event log est excellent pour écrire, mais inconvenient pour les requêtes du type « lister les commandes payées du client C7 ». Projection résout ce problème. C’est un consumer qui lit les événements et maintient un modèle optimisé pour la lecture. C’est exactement la partie « Q » dans CQRS.

public record OrderSummary(String orderId, String customerId, String status, BigDecimal total) {}

@Component
public class OrderSummaryProjection {

    private final Map<String, OrderSummary> summaries = new ConcurrentHashMap<>();

    @KafkaListener(topics = "order-events", groupId = "order-summary-projection")
    void on(OrderEvent event) {
        summaries.compute(event.orderId(), (id, current) -> switch (event) {
            case OrderEvent.Created e -> new OrderSummary(id, e.customerId(), "NEW", BigDecimal.ZERO);
            case OrderEvent.ItemAdded e -> new OrderSummary(id, current.customerId(), current.status(),
                    current.total().add(e.price()));
            case OrderEvent.Paid e -> new OrderSummary(id, current.customerId(), "PAID", current.total());
            case OrderEvent.Cancelled e -> new OrderSummary(id, current.customerId(), "CANCELLED", current.total());
        });
    }

    public List<OrderSummary> paidOrdersOf(String customerId) {
        return summaries.values().stream()
                .filter(s -> s.customerId().equals(customerId) && s.status().equals("PAID"))
                .toList();
    }
}

Cet exemple utilise ConcurrentHashMap pour plus de concision. En réalité, vous écrivez dans PostgreSQL, Elasticsearch ou Redis. Le point important est que la projection a un groupId fixe, donc Kafka se souvient de l’offset et ne traite que les nouveaux événements après chaque redémarrage.

La projection doit être idempotente. Kafka garantit at-least-once, donc un événement peut être livré à nouveau après un rebalance. Avec une base de données, vous pouvez enregistrer le dernier offset traité par partition et ignorer les enregistrements plus anciens.

7.1. Replay pour reconstruire le read model

Supposons que la projection ait un bug qui fait que total est mal calculé. Vous corrigez le code, supprimez les données du read model, arrêtez l’application puis réinitialisez l’offset du group au début :

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-summary-projection \
  --topic order-events \
  --reset-offsets --to-earliest --execute

GROUP                     TOPIC         PARTITION  NEW-OFFSET
order-summary-projection  order-events  0          0
order-summary-projection  order-events  1          0
order-summary-projection  order-events  2          0

Au redémarrage, la projection relit tout l’historique et crée le bon read model. Pour ajouter un rapport complètement nouveau, vous écrivez simplement un nouveau listener avec un nouveau groupId. Ce rapport a automatiquement toutes les données depuis le premier jour. C’est le plus grand avantage de stocker les événements plutôt que l’état.

8. Erreurs courantes et quand ne pas l’utiliser

Changer le schéma des événements. Les anciens événements existent à jamais. Ajoutez seulement de nouveaux champs avec une valeur par défaut, et ne supprimez ni ne changez le sens des anciens champs. Pour les changements importants, créez un nouveau type d’événement (ItemAddedV2) ou écrivez un « upcaster » pour convertir les anciens événements lors de la lecture. Pour les grands systèmes, utilisez Schema Registry avec Avro ou Protobuf.

Fourrager l’état dans les événements. Un événement du type OrderUpdated { toute la commande } n’est que du CRUD déguisé en event sourcing. Un événement doit décrire une intention métier, comme ItemAdded ou OrderCancelled.

Replay trop long. Un aggregate avec des milliers d’événements se chargera lentement. La solution est un snapshot : enregistrez régulièrement l’état avec version dans un topic compacté, au chargement lisez le snapshot puis rejouez seulement les événements après cela.

Appeler des side effects dans apply. Si apply envoie un email, chaque replay renverra l’email à nouveau. apply ne peut que changer l’état en mémoire.

Données personnelles et droit à l’oubli. Le log immuable entre en conflit avec la demande de suppression des données utilisateur. La technique couramment utilisée est crypto-shredding : chiffrez les données personnelles avec une clé spécifique pour chaque client, quand vous avez besoin de supprimer, détruisez la clé.

Vous ne devriez pas utiliser event sourcing quand :

Le domaine est principalement du CRUD simple et personne n’a besoin de l’historique.
L’équipe n’est pas familière avec l’eventual consistency entre le côté écriture et le read model.
Vous avez besoin de requêtes cohérentes immédiatement après l’écriture sur plusieurs agrégats.

Remarque : Kafka est un excellent event log, mais pas un event store complet. Il manque deux choses : la lecture rapide par aggregate et l’écriture conditionnelle selon la version. De nombreux systèmes utilisent une base de données ou un event store dédié comme source de vérité, puis utilisent Kafka (généralement via le outbox pattern) pour distribuer les événements aux projections et autres services.

9. Conclusion

Dans cet article, vous avez enregistré les commandes comme une séquence d’événements immuables sur Kafka au lieu d’un enregistrement écrasé. Vous avez modélisé les événements avec sealed interface et record, reconstruit l’état avec Order.replay et vérifié les règles métier avant d’ajouter. Vous avez également construit le read model par projection et l’avez reconstruit en réinitialisant l’offset.

Il y a trois points à retenir :

Utilisez orderId comme clé pour préserver l’ordre des événements de chaque commande.
Désactivez la rétention basée sur le temps et n’utilisez pas de compaction pour le topic d’événements.
Comprenez bien les limites de concurrency de Kafka avant de le considérer comme votre unique source de vérité.

Quand elle est appliquée au bon endroit, l’event sourcing vous donne un historique complet et la capacité de créer de nouveaux read models sans migration.