Saga-Pattern mit Kafka für verteilte Transaktionen: Tourenreservierung

1. Übersicht

Wenn ein Geschäftsprozess mehrere Microservices durchläuft und jeder Service seine eigene Datenbank hat, kannst du nicht alles in ein @Transactional wrappen. Das Saga-Pattern löst dieses Problem. Statt einer großen Transaktion führst du eine Reihe von lokalen Transaktionen durch. Jeder Schritt hat eine Kompensationsmaßnahme, um rückgängig zu machen, wenn der nächste Schritt fehlschlägt.

In diesem Artikel implementierst du eine Saga mit Orchestration in Spring Boot 3 und Kafka für den Geschäftsprozess Tourenreservierung. Nach dieser Anleitung wirst du einen Orchestrator haben, der die Schritte koordiniert, Participant-Services, die über Kafka antworten, und einen Test für den Kompensationsfluss.

2. Das Problem: Tourenreservierung

Eine Tourenreservierung besteht aus drei Schritten, wobei jeder Schritt zu einem Service gehört:

FlightService: ein Flugticket reservieren.
HotelService: ein Hotelzimmer reservieren.
PaymentService: den Kunden belasten.

Angenommen, Flugticket und Hotelzimmer sind bereits reserviert, aber die Zahlung fehlschlägt, weil die Kartenlimits erreicht sind. Dann muss das System das Hotelzimmer und dann die Flugbuchung stornieren. Sonst bezahlt der Kunde nicht, sondern hält die Reservierung.

Das Zwei-Phasen-Commit-Protokoll (2PC) könnte das lösen, aber es blockiert Ressourcen während der gesamten Transaktion. Es erfordert auch, dass jeder Service XA unterstützt, und Kafka kann nicht an XA teilnehmen. Saga akzeptiert eventuelle Konsistenz: die Daten können vorübergehend inkonsistent sein, werden aber schließlich konsistent sein.

3. Choreography oder Orchestration

Sagas können auf zwei Arten organisiert werden:

Kriterium	Choreography	Orchestration
Orchestrierung	Jeder Service hört auf Events von anderen Services	Ein Orchestrator sendet Befehle und erhält Antworten
Geschäftsprozessfluss	Verteilt auf mehrere Services	An einem Ort konzentriert
Geeignet für	2–3 Schritte, wenig Änderungen	Viele Schritte, mit komplexer Kompensationslogik
Risiko	Zirkuläre Abhängigkeiten, schwer zu debuggen	Orchestrator wird ein zentraler Ausfallpunkt für die Logik

Mit drei Schritten und einem umgekehrten Kompensationsfluss ist Orchestration leichter zu lesen und zu testen. Dieser Artikel verwendet diesen Ansatz. Die Kafka-Topics sind wie folgt organisiert:

tour.commands.flight   ← Orchestrator sendet Befehle an FlightService
tour.commands.hotel    ← Orchestrator sendet Befehle an HotelService
tour.commands.payment  ← Orchestrator sendet Befehle an PaymentService
tour.replies           ← alle Services senden Ergebnisse an den Orchestrator

4. Projekteinrichtung

4.1. Abhängigkeiten

Du musst Spring for Apache Kafka hinzufügen. Spring Boot verwaltet die Version für dich:

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

Nachrichten werden als JSON versendet. Konfiguriere den Serializer und Deserializer wie folgt:

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 schreibt den Klassennamen in den Header __TypeId__. Auf der Consumer-Seite liest JsonDeserializer diesen Header, um zu wissen, welches Objekt erstellt werden soll. trusted.packages beschränkt die Pakete, die deserialisiert werden dürfen, um Sicherheitslücken zu vermeiden.

Als nächstes deklarierst du die 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. Message-Modell

Du verwendest Java-17-Records, um Commands und Replies darzustellen:

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) {}

Jede Nachricht trägt eine bookingId. Dieser Wert ist gleichzeitig die Saga-Kennung und der Message Key, wenn die Nachricht zu Kafka gesendet wird. Dadurch fallen alle Nachrichten derselben Buchung in dieselbe Partition und werden in der richtigen Reihenfolge verarbeitet.

6. Definition der Saga als State Machine

Der wichtigste Teil ist die Logik, die den nächsten Schritt bestimmt. Du solltest diese Logik von Kafka trennen, um sie mit reinen Unit Tests testen zu können.

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 ist ein Stack, der die bereits erfolgreich durchgeführten Schritte speichert. Wenn du kompensieren musst, pop du vom Stack, sodass die Schritte in umgekehrter Reihenfolge rückgängig gemacht werden.

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(); // die Saga ist abgeschlossen, doppelte Antwort ignorieren
        }
        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)); // versuche erneut zu stornieren
        }
        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());
    }
}

Die Methode onReply behandelt vier Fälle:

RESERVE erfolgreich: registriert diesen Schritt im Stack. Wenn dies der letzte Schritt ist, wechselt die Saga zu COMPLETED. Wenn nicht, sendet die Saga den RESERVE-Befehl für den nächsten Schritt.
RESERVE fehlgeschlagen: wechselt die Saga zu COMPENSATING und beginnt, den zuletzt erfolgreich durchgeführten Schritt zu stornieren.
CANCEL erfolgreich: storniert die Saga den vorherigen Schritt. Wenn der Stack leer ist, wechselt die Saga zu CANCELLED.
CANCEL fehlgeschlagen: sendet die Saga den Stornierungsbefehl erneut. Die Kompensation darf nicht abgebrochen werden.

Beachte, dass der fehlgeschlagene Schritt nicht im Stack ist. Wenn die Zahlung fehlschlägt, gibt es nichts zu erstatten, muss die Saga also nur das Hotel und das Flugticket stornieren.

7. Orchestrator

Der Orchestrator macht zwei Dinge: sendet Befehle an Kafka und hört auf Antworten vom topic 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() verwendet bookingId als Schlüssel. Der Topic-Name wird aus dem Step abgeleitet, z.B. wird HOTEL zu tour.commands.hotel.

Anmerkung: ConcurrentHashMap ist nur für das Demo gedacht. In der Produktion muss der Saga-Status in einer Datenbank gespeichert werden. Wenn der Orchestrator mitten drin neu startet, muss er den Status erneut lesen, um fortzufahren oder zu kompensieren.

8. Participant-Service

Jeder Participant hört auf seinen Command-Topic, führt eine lokale Transaktion durch und sendet dann eine Antwort. Beispiel: 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 : "Keine Zimmer verfügbar"));
    }

    private boolean reserve(String bookingId) {
        if (reservedRooms.contains(bookingId)) {
            return true; // bereits reserviert, Message wurde erneut gesendet
        }
        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; // Stornierung von nie reserviertem Zimmer ist immer erfolgreich
    }
}

Zwei wichtige Details im obigen Code:

reserve() ist idempotent. Kafka garantiert mindestens eine Zustellung, daher kann derselbe Befehl zweimal ankommen. Wenn bereits für diese bookingId ein Zimmer reserviert ist, gibt der Handler den Erfolg zurück, ohne das Zimmer erneut zu verringern.
cancel() ist ebenfalls idempotent und immer erfolgreich. Der Stornierungsbefehl kann vor dem Reservierungsbefehl ankommen oder für eine Buchung ankommen, bei der kein Zimmer reserviert wurde. Beide Fälle dürfen die Daten nicht beschädigen.

FlightCommandHandler und PaymentCommandHandler haben eine ähnliche Struktur. Bei Payment bedeutet CANCEL, das Geld zurückzuerstatten (refund).

Wenn PaymentService die Zahlung ablehnt, sieht das Orchestrator-Logfile wie folgt aus:

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

Das Zimmer wird zuerst storniert, das Flugticket dann, in umgekehrter Reihenfolge wie bei der Reservierung.

9. Saga-Flow testen

Da TourBookingSaga nicht von Kafka abhängt, kannst du es mit JUnit 5 und AssertJ testen, ohne einen Broker zu brauchen:

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, "Karte wurde abgelehnt")).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);
    }
}

Der zweite Test überprüft das wichtigste Verhalten einer Saga. Wenn die Zahlung fehlschlägt, müssen das Hotel und das Flugticket storniert werden, und die Saga muss bei CANCELLED enden. Wenn du End-to-End-Tests brauchst, kannst du @EmbeddedKafka aus spring-kafka-test verwenden oder Testcontainers. Die Logik sollte auf Unit-Test-Ebene wie oben getestet werden.

10. Häufige Fehler

Dual write. Der Participant schreibt zuerst in die Datenbank und sendet dann die Antwort. Wenn der Service zwischen den beiden Operationen abstürzt, bleibt die Saga stecken. Die Standardlösung ist Transactional Outbox: schreibe die Antwort in die outbox-Tabelle in derselben Transaktion wie die Geschäftsdaten. Danach schiebt ein separater Prozess (oder Debezium CDC) diese Datensätze zu Kafka.

Fehlende Idempotenz. Consumer-Rebalance oder Retry sendet die Nachricht erneut. Jeder Handler, einschließlich Kompensationshandler, muss Duplikate sicher verarbeiten. Im Orchestrator solltest du speichern, welcher Schritt auf eine Antwort wartet, und Antworten ignorieren, die nicht passen.

Keinen Schlüssel verwenden. Wenn du eine Nachricht ohne Schlüssel sendest, können Nachrichten derselben Buchung in verschiedene Partitionen fallen. Dann könnte der CANCEL-Befehl vor dem RESERVE-Befehl verarbeitet werden.

Fehlende Isolierung. Sagas haben nicht das ACID-Isolations-Merkmal. Während die Saga läuft, können andere Services das Zimmer als ‘reserviert’ sehen, obwohl es später storniert werden könnte. Du solltest einen Zwischenstatus wie PENDING (semantic lock) verwenden und wechsle nur zu CONFIRMED, wenn die Saga abgeschlossen ist.

Fehlendes Timeout. Wenn ein Service niemals antwortet, wartet die Saga auf ewig. Du brauchst einen regelmäßigen Job, der abgelaufene Sagas scannt, um sie zu kompensieren. Wenn die CANCEL-Befehle immer wieder fehlschlagen, solltest du die Anzahl der Wiederholungen begrenzen und die Nachricht in ein dead-letter-Topic verschieben zur manuellen Verarbeitung.

Anmerkung: Verwende Saga nicht, wenn alle Daten in derselben Datenbank sind. Ein normales @Transactional ist viel einfacher und sicherer. Saga ist auch nicht geeignet für Geschäftsprozesse, die keinen Zwischenstatus akzeptieren, oder Schritte, die nicht kompensiert werden können. Zum Beispiel kann eine versendete Bestätigungs-E-Mail nicht ‘zurückgerufen’ werden. Du solltest solche Schritte am Ende der Saga platzieren.

11. Fazit

In diesem Artikel hast du das Saga-Pattern mit Orchestration und Kafka für den Geschäftsprozess Tourenreservierung implementiert:

Die Saga-Logik ist in eine reine State Machine getrennt, testbar, ohne einen Broker zu benötigen.
Der Orchestrator sendet Befehle und empfängt Antworten über Kafka, verwendet bookingId als Schlüssel, um die Reihenfolge beizubehalten.
Kompensationsschritte werden in umgekehrter Reihenfolge durchgeführt, dank eines Stacks.
Participants verarbeiten Befehle idempotent.

Wenn du die Saga in die Produktion bringst, musst du drei Dinge hinzufügen: speichere den Saga-Status in einer Datenbank, verwende Transactional Outbox, um Dual-Write zu vermeiden, und habe einen Timeout-Mechanismus und ein dead-letter-Topic. Dies sind die Teile, die bestimmen, ob deine Saga wirklich zuverlässig ist oder nicht.