Saga pattern avec Kafka : transactions distribuées et réservation

1. Aperçu

Quand un processus métier traverse plusieurs microservices, chacun avec sa propre base de données, vous ne pouvez pas envelopper le tout dans un @Transactional. Le pattern Saga résout ce problème. Au lieu d’une grande transaction, vous exécutez une suite de transactions locales. Chaque étape possède une action de compensation pour annuler si l’étape suivante échoue.

Dans cet article, vous allez implémenter une saga en style orchestration avec Spring Boot 3 et Kafka pour le processus de réservation de tour. Après votre lecture, vous aurez un orchestrator qui coordonne les étapes, des services participants qui répondent via Kafka, et une suite de tests pour le flux de compensation.

2. Le problème : réservation de tour

Une réservation de tour comporte trois étapes, chacune dans un service différent :

1. FlightService : réserver un vol.
2. HotelService : réserver une chambre d’hôtel.
3. PaymentService : débiter le client.

Supposons que le vol et la chambre sont réservés, mais le paiement échoue parce que la carte n’a pas assez de limite. Dans ce cas, le système doit annuler la chambre puis le vol. Sinon, le client garderait la réservation sans payer.

Le two-phase commit (2PC) peut résoudre cela, mais il verrouille les ressources pendant toute la transaction. Il nécessite aussi que chaque service supporte XA, et Kafka ne peut pas participer à XA. Saga accepte la eventual consistency : les données peuvent être temporairement incohérentes, mais finiront par converger.

3. Choreography ou Orchestration

Une saga s’organise de deux façons :

Critère	Choreography	Orchestration
Coordination	Chaque service écoute les événements des autres	Un orchestrator envoie des commandes et reçoit des réponses
Flux métier	Éparpillé sur plusieurs services	Centralisé au même endroit
Adapté pour	2–3 étapes, peu de changements	Plusieurs étapes, logique de compensation complexe
Risque	Dépendances circulaires, difficile à déboguer	L’orchestrator devient un point unique de logique

Avec trois étapes et un flux de compensation qui remonte en arrière, l’orchestration est plus facile à lire et à tester. Cet article utilise cette approche. Les topics Kafka sont organisés ainsi :

tour.commands.flight   ← orchestrator envoie des ordres à FlightService
tour.commands.hotel    ← orchestrator envoie des ordres à HotelService
tour.commands.payment  ← orchestrator envoie des ordres à PaymentService
tour.replies           ← tous les services renvoient les résultats à l'orchestrator

4. Configuration du projet

4.1. Dépendances

Vous avez besoin d’ajouter Spring for Apache Kafka. Spring Boot gère les versions pour vous :

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

4.2. Configuration

Les messages sont envoyés en JSON. Configurez les sérialiseurs et désérialiseurs comme suit :

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      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.tour

JsonSerializer écrit le nom de la classe dans l’en-tête __TypeId__. Côté consumer, JsonDeserializer lit cet en-tête pour savoir quel objet créer. trusted.packages restreint les packages autorisés à être désérialisés, pour éviter les failles de sécurité.

Ensuite, déclarez les topics :

@Configuration
public class KafkaTopicConfig {

    @Bean
    KafkaAdmin.NewTopics sagaTopics() {
        return new KafkaAdmin.NewTopics(
            TopicBuilder.name("tour.commands.flight").partitions(3).build(),
            TopicBuilder.name("tour.commands.hotel").partitions(3).build(),
            TopicBuilder.name("tour.commands.payment").partitions(3).build(),
            TopicBuilder.name("tour.replies").partitions(3).build());
    }
}

5. Modèle de message

Vous utilisez les record de Java 21 pour représenter les commandes et réponses :

package com.nonla.tour;

public enum Step { FLIGHT, HOTEL, PAYMENT }

public enum Action { RESERVE, CANCEL }

public record SagaCommand(String bookingId, Step step, Action action, BigDecimal amount) {}

public record SagaReply(String bookingId, Step step, Action action,
                        boolean success, String reason) {}

Chaque message porte un bookingId. Cette valeur est à la fois l’identifiant de la saga et la clé du message quand envoyé à Kafka. Grâce à cela, tous les messages d’une même réservation tombent dans la même partition et sont traités dans le bon ordre.

6. Définir la saga comme une machine à états

La partie la plus importante est la logique qui décide l’étape suivante. Vous devriez isoler cette logique de Kafka pour la tester avec un test unitaire pur.

public enum SagaStatus { RUNNING, COMPENSATING, COMPLETED, CANCELLED }

public class SagaState {
    private final String bookingId;
    private final BigDecimal amount;
    private final Deque<Step> completed = new ArrayDeque<>();
    private SagaStatus status = SagaStatus.RUNNING;

    public SagaState(String bookingId, BigDecimal amount) {
        this.bookingId = bookingId;
        this.amount = amount;
    }

    public String bookingId() { return bookingId; }
    public BigDecimal amount() { return amount; }
    public Deque<Step> completed() { return completed; }
    public SagaStatus status() { return status; }
    public void status(SagaStatus status) { this.status = status; }
}

completed est une pile mémorisant les étapes réussies. Quand il faut compenser, vous dépiliez du sommet, donc les étapes sont annulées dans l’ordre inverse.

public class TourBookingSaga {

    static final List<Step> STEPS = List.of(Step.FLIGHT, Step.HOTEL, Step.PAYMENT);

    public Optional<SagaCommand> start(SagaState s) {
        return Optional.of(command(s, STEPS.getFirst(), Action.RESERVE));
    }

    public Optional<SagaCommand> onReply(SagaState s, SagaReply r) {
        if (s.status() == SagaStatus.COMPLETED || s.status() == SagaStatus.CANCELLED) {
            return Optional.empty(); // saga terminée, ignorer la réponse en double
        }
        if (r.action() == Action.RESERVE && r.success()) {
            s.completed().push(r.step());
            int idx = STEPS.indexOf(r.step());
            if (idx == STEPS.size() - 1) {
                s.status(SagaStatus.COMPLETED);
                return Optional.empty();
            }
            return Optional.of(command(s, STEPS.get(idx + 1), Action.RESERVE));
        }
        if (r.action() == Action.CANCEL && !r.success()) {
            return Optional.of(command(s, r.step(), Action.CANCEL)); // réessayer d'annuler
        }
        s.status(SagaStatus.COMPENSATING);
        return compensateNext(s);
    }

    private Optional<SagaCommand> compensateNext(SagaState s) {
        if (s.completed().isEmpty()) {
            s.status(SagaStatus.CANCELLED);
            return Optional.empty();
        }
        return Optional.of(command(s, s.completed().pop(), Action.CANCEL));
    }

    private SagaCommand command(SagaState s, Step step, Action action) {
        return new SagaCommand(s.bookingId(), step, action, s.amount());
    }
}

La méthode onReply traite quatre cas :

• RESERVE succès : enregistrer cette étape dans la pile. Si c’est la dernière étape, la saga passe à COMPLETED. Sinon, envoyer une commande RESERVE pour l’étape suivante.
• RESERVE échec : la saga passe à COMPENSATING et commence à annuler l’étape la plus récemment réussie.
• CANCEL succès : la saga annule l’étape précédente. Quand la pile est vide, la saga passe à CANCELLED.
• CANCEL échec : la saga renvoie la commande d’annulation. La compensation ne peut pas être abandonnée.

Notez que l’étape échouée n’est pas dans la pile. Si le paiement échoue, il n’y a rien à rembourser, donc la saga doit seulement annuler la chambre et le vol.

7. L’orchestrator

L’orchestrator fait deux choses : envoyer les commandes à Kafka et écouter les réponses de tour.replies.

@Service
public class TourBookingOrchestrator {

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

    private final KafkaTemplate<String, Object> kafka;
    private final TourBookingSaga saga = new TourBookingSaga();
    private final Map<String, SagaState> store = new ConcurrentHashMap<>();

    public TourBookingOrchestrator(KafkaTemplate<String, Object> kafka) {
        this.kafka = kafka;
    }

    public void book(String bookingId, BigDecimal amount) {
        SagaState state = new SagaState(bookingId, amount);
        store.put(bookingId, state);
        saga.start(state).ifPresent(this::send);
    }

    @KafkaListener(topics = "tour.replies", groupId = "tour-orchestrator")
    public void onReply(SagaReply reply) {
        SagaState state = store.get(reply.bookingId());
        if (state == null) {
            return;
        }
        saga.onReply(state, reply).ifPresent(this::send);
        log.info("booking={} reply={}/{} ok={} -> status={}",
            reply.bookingId(), reply.step(), reply.action(), reply.success(), state.status());
    }

    private void send(SagaCommand cmd) {
        String topic = "tour.commands." + cmd.step().name().toLowerCase();
        kafka.send(topic, cmd.bookingId(), cmd);
    }

    public SagaStatus statusOf(String bookingId) {
        return store.get(bookingId).status();
    }
}

send() utilise bookingId comme clé. Le nom du topic est dérivé de Step, par exemple HOTEL devient tour.commands.hotel.

Attention : ConcurrentHashMap n’est que pour la démo. En production, l’état de la saga doit être persisté dans une base de données. Si l’orchestrator redémarre en cours de route, il doit relire l’état pour continuer ou compenser.

8. Service participant

Chaque participant écoute le topic de commande qui lui est destiné, effectue une transaction locale, puis envoie une réponse. Exemple avec HotelService :

@Component
public class HotelCommandHandler {

    private final KafkaTemplate<String, Object> kafka;
    private final Set<String> reservedRooms = ConcurrentHashMap.newKeySet();
    private final AtomicInteger availableRooms = new AtomicInteger(10);

    public HotelCommandHandler(KafkaTemplate<String, Object> kafka) {
        this.kafka = kafka;
    }

    @KafkaListener(topics = "tour.commands.hotel", groupId = "hotel-service")
    public void handle(SagaCommand cmd) {
        boolean ok = switch (cmd.action()) {
            case RESERVE -> reserve(cmd.bookingId());
            case CANCEL -> cancel(cmd.bookingId());
        };
        kafka.send("tour.replies", cmd.bookingId(),
            new SagaReply(cmd.bookingId(), Step.HOTEL, cmd.action(), ok, ok ? null : "Pas de chambre disponible"));
    }

    private boolean reserve(String bookingId) {
        if (reservedRooms.contains(bookingId)) {
            return true; // déjà réservée, message renvoyé en doublon
        }
        if (availableRooms.getAndUpdate(n -> n > 0 ? n - 1 : n) == 0) {
            return false;
        }
        reservedRooms.add(bookingId);
        return true;
    }

    private boolean cancel(String bookingId) {
        if (reservedRooms.remove(bookingId)) {
            availableRooms.incrementAndGet();
        }
        return true; // annuler ce qui n'a jamais été réservé c'est un succès
    }
}

Deux détails importants dans ce code :

• reserve() est idempotent. Kafka garantit la livraison at-least-once, donc le même message peut arriver deux fois. Si la chambre est déjà réservée pour ce bookingId, le handler retourne succès sans décrémenter le compteur.
• cancel() est aussi idempotent et toujours un succès. La commande d’annulation peut arriver avant la commande de réservation, ou pour une réservation qui n’a jamais existé. Les deux cas ne doivent pas corrompre les données.

FlightCommandHandler et PaymentCommandHandler ont une structure similaire. Avec Payment, CANCEL signifie rembourser.

Quand PaymentService refuse le paiement, le log de l’orchestrator ressemble à :

booking=BK-1001 reply=FLIGHT/RESERVE ok=true -> status=RUNNING
booking=BK-1001 reply=HOTEL/RESERVE ok=true -> status=RUNNING
booking=BK-1001 reply=PAYMENT/RESERVE ok=false -> status=COMPENSATING
booking=BK-1001 reply=HOTEL/CANCEL ok=true -> status=COMPENSATING
booking=BK-1001 reply=FLIGHT/CANCEL ok=true -> status=CANCELLED

La chambre est annulée en premier, le vol en dernier, dans l’ordre inverse de la réservation.

9. Test du flux saga

Puisque TourBookingSaga ne dépend pas de Kafka, vous la testez avec JUnit 5 et AssertJ sans besoin de broker :

class TourBookingSagaTest {

    private final TourBookingSaga saga = new TourBookingSaga();

    private SagaReply ok(Step step, Action action) {
        return new SagaReply("BK-1", step, action, true, null);
    }

    @Test
    void whenAllStepsSucceed_thenSagaCompleted() {
        SagaState s = new SagaState("BK-1", new BigDecimal("15000000"));

        assertThat(saga.start(s)).get().extracting(SagaCommand::step).isEqualTo(Step.FLIGHT);
        assertThat(saga.onReply(s, ok(Step.FLIGHT, Action.RESERVE)))
            .get().extracting(SagaCommand::step).isEqualTo(Step.HOTEL);
        assertThat(saga.onReply(s, ok(Step.HOTEL, Action.RESERVE)))
            .get().extracting(SagaCommand::step).isEqualTo(Step.PAYMENT);
        assertThat(saga.onReply(s, ok(Step.PAYMENT, Action.RESERVE))).isEmpty();

        assertThat(s.status()).isEqualTo(SagaStatus.COMPLETED);
    }

    @Test
    void whenPaymentFails_thenCompensateHotelThenFlight() {
        SagaState s = new SagaState("BK-1", new BigDecimal("15000000"));
        saga.start(s);
        saga.onReply(s, ok(Step.FLIGHT, Action.RESERVE));
        saga.onReply(s, ok(Step.HOTEL, Action.RESERVE));

        SagaCommand first = saga.onReply(s,
            new SagaReply("BK-1", Step.PAYMENT, Action.RESERVE, false, "Carte refusée")).orElseThrow();
        assertThat(first).extracting(SagaCommand::step, SagaCommand::action)
            .containsExactly(Step.HOTEL, Action.CANCEL);
        assertThat(s.status()).isEqualTo(SagaStatus.COMPENSATING);

        SagaCommand second = saga.onReply(s, ok(Step.HOTEL, Action.CANCEL)).orElseThrow();
        assertThat(second).extracting(SagaCommand::step, SagaCommand::action)
            .containsExactly(Step.FLIGHT, Action.CANCEL);

        assertThat(saga.onReply(s, ok(Step.FLIGHT, Action.CANCEL))).isEmpty();
        assertThat(s.status()).isEqualTo(SagaStatus.CANCELLED);
    }
}

Le deuxième test vérifie le comportement le plus important de la saga. Un paiement échoué doit annuler l’hôtel puis le vol, et la saga doit finir en CANCELLED. Pour les tests end-to-end, vous pouvez utiliser @EmbeddedKafka de spring-kafka-test ou Testcontainers. La logique se teste mieux au niveau unitaire comme ci-dessus.

10. Erreurs courantes

Dual write. Le participant écrit en base de données puis envoie la réponse. Si le service meurt entre les deux, la saga est figée. La solution correcte est le Transactional Outbox : écrire la réponse dans une table outbox dans la même transaction que les données métier. Ensuite un processus séparé (ou Debezium CDC) pousse ces enregistrements dans Kafka.

Pas d’idempotence. Le rééquilibrage du consumer ou une retry renverra le message. Tous les handlers, y compris les handlers de compensation, doivent traiter les doublons en toute sécurité. À l’orchestrator, vous devriez sauvegarder aussi l’étape en attente de réponse et ignorer les réponses qui ne correspondent pas.

Pas de clé. Envoyer un message sans clé risque que plusieurs messages d’une même réservation tombent dans différentes partitions. Une commande CANCEL pourrait alors être traitée avant RESERVE.

Isolation insuffisante. La saga n’a pas l’isolement (Isolation) comme ACID. Pendant l’exécution, un autre service peut voir la chambre “réservée” même si elle sera annulée plus tard. Vous devriez utiliser un état intermédiaire comme PENDING (semantic lock) et ne passer à CONFIRMED que quand la saga est terminée.

Pas de timeout. Si un service ne répond jamais, la saga attend à l’infini. Vous avez besoin d’un job qui scanne régulièrement les sagas expirées pour les compenser. Si les commandes CANCEL échouent de façon répétée, limitez le nombre de tentatives et envoyez le message dans une dead-letter topic pour une intervention humaine.

Rappel : N’utilisez pas Saga si toutes les données sont dans la même base. Un @Transactional ordinaire est bien plus simple et sûr. Saga ne convient pas non plus aux métiers qui ne tolèrent pas les états intermédiaires, ou qui ont des étapes non réversibles. Par exemple, un email de confirmation une fois envoyé ne peut pas être “rappelé”. Vous devriez mettre ce type d’étapes à la fin de la saga.

11. Conclusion

Dans cet article, vous avez implémenté le pattern Saga en orchestration avec Kafka pour la réservation de tour :

• La logique saga est isolée en une machine à états pure, testable sans broker.
• L’orchestrator envoie des commandes et reçoit les réponses via Kafka, en utilisant bookingId comme clé pour préserver l’ordre.
• Les étapes de compensation sont exécutées dans l’ordre inverse grâce à une pile.
• Les participants traitent les commandes de façon idempotente.

Quand vous mettez la saga en production, vous devez ajouter trois choses : persister l’état de la saga dans une base de données, utiliser Transactional Outbox pour éviter le dual write, et avoir un mécanisme de timeout avec dead-letter topic. Ces éléments détermineront si votre saga est vraiment fiable en pratique.