Event Sourcing mit Kafka: Ereignisse speichern statt Status

1. Übersicht

Die meisten Anwendungen speichern nur den aktuellen Status. Wenn eine Bestellung storniert wird, ändert sich das Feld status in der Tabelle orders auf CANCELLED. Niemand weiß danach noch, welche Artikel ursprünglich in der Bestellung waren und ob schon bezahlt wurde.

Event Sourcing funktioniert umgekehrt: Du speicherst nur die Abfolge von Ereignissen, die tatsächlich vorgefallen sind, und der Status ist das Ergebnis dieser Abfolge. In diesem Artikel wirst du Kafka als Event Log für das Aggregate Order mit Spring Boot 3.x nutzen. Du wirst den Status durch Replay von Ereignissen rekonstruieren, ein Read Model durch Projektion aufbauen und die praktischen Grenzen dieses Ansatzes verstehen lernen.

2. Status speichern vs. Ereignisse speichern

Bei der traditionellen Methode ist eine Bestellung ein einzelner Datensatz:

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

Mit Event Sourcing ist dieselbe Bestellung eine Abfolge von Ereignissen, die nur hinzugefügt werden, nie geändert oder gelöscht:

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="Kunde hat sich umentschieden")

Der Status CANCELLED, total=250000 wird berechnet durch “Falten” (fold) der Ereignisse in Reihenfolge. Diese Speicherweise bringt drei Vorteile:

Natürliches Audit: Die vollständige Historie ist die Quelldaten selbst.
Neue Read Models benötigen keine Migration: Wenn du einen neuen Report hinzufügen möchtest, kannst du das Log in eine neue Projektion erneut verarbeiten.
Zeitpunkt-Debugging: Du kannst den Status zu jedem beliebigen Zeitpunkt rekonstruieren.

Kafka eignet sich hervorragend für diese Log-Rolle. Ein Topic ist eine append-only Sequenz mit garantierter Reihenfolge in jeder Partition. Wenn du die orderId als Schlüssel verwendest, fallen alle Ereignisse derselben Bestellung in dieselbe Partition und behalten die richtige Reihenfolge.

3. Projekt-Setup

3.1. Dependencies

<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 verwaltet die Version von spring-kafka, daher musst du <version> nicht angeben. spring-boot-starter-json enthält Jackson und das jsr310-Modul zum Serialisieren von Instant.

3.2. Konfiguration

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

Wichtige Punkte:

acks: all und enable.idempotence: true garantieren, dass Ereignisse dauerhaft geschrieben werden und Producer-Retries keine Duplikate erzeugen.
Wir deaktivieren die Type-Header von Spring. Der Ereignistyp ist direkt im JSON über das Attribut type enthalten, damit Consumer in anderen Sprachen ihn trotzdem lesen können.
auto-offset-reset: earliest erlaubt einer neuen Consumer-Gruppe, das gesamte Log von vorne zu lesen. Das ist eine Voraussetzung für Replay.

3.3. Topic für dauerhafte Speicherung von Ereignissen

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

Standardmäßig löscht Kafka Daten nach 7 Tagen. Bei einem Event Store bedeutet das den Verlust der Quelldaten. retention.ms=-1 deaktiviert das zeitbasierte Löschen.

Hinweis: Verwende nicht cleanup.policy=compact für Ereignis-Topics. Compaction behält nur den letzten Datensatz pro Schlüssel, löscht also die gesamte Historie einer Bestellung. Compaction ist nur für Snapshot-Topics geeignet.

4. Ereignisse definieren

Ereignisse sind unveränderliche Daten, die beschreiben, was vorgefallen ist und im Perfekt benannt sind. sealed interface zusammen mit record von Java 21 eignet sich hervorragend dafür:

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

Wenn du ein ItemAdded sendest, hat die Payload auf Kafka folgende Form:

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

Der Name in @JsonSubTypes ist ein langfristiger Vertrag. Du kannst den Java-Klassennamen ändern, aber nicht die Zeichenkette "ItemAdded", sonst können alte Ereignisse nicht mehr gelesen werden.

5. Aggregate: Status aus Ereignissen aufbauen

Order hat zwei Arten von Methoden:

Commands (addItem, pay, cancel): prüfen Geschäftsregeln und geben Ereignisse zurück. Commands ändern nicht selbst den Status.
apply: nimmt ein Ereignis und aktualisiert den Status. Diese Funktion darf keine Geschäftsfehler werfen, denn das Ereignis ist bereits vorgefallen.

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, "Artikel können nur zu neuen Bestellungen hinzugefügt werden");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "Bestellung befindet sich nicht im Zahlungsausstehend-Status");
        if (total.signum() == 0) {
            throw new IllegalStateException("Leere Bestellung kann nicht bezahlt werden");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("Bestellung ist bereits storniert");
        }
        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; }
}

Da OrderEvent sealed ist, überprüft der Compiler das switch in apply vollständig. Wenn du einen neuen Ereignistyp hinzufügst und vergisst, ihn zu behandeln, wird der Code nicht kompiliert. Das ist ein wertvolleres Sicherheitsnetz.

5.1. Aggregate testen

Das Aggregate hängt nicht von Kafka ab, daher sind Tests sehr schnell:

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", "Kunde hat sich umentschieden", 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", "Artikel nicht verfügbar", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("Zahlungsausstehend");
    }
}

Der erste Test zeigt, dass vier Ereignisse zu einem konsistenten Status zusammengefasst werden, mit version gleich der Anzahl der angewendeten Ereignisse. Der zweite Test zeigt, dass Geschäftsregeln anhand des aus der Historie rekonstruierten Status geprüft werden.

6. Ereignisse in Kafka schreiben

6.1. Event Store

Um Commands zu verarbeiten, brauchst du die Historie einer Bestellung. Kafka unterstützt keine Anfragen wie “gib mir alle Datensätze mit Schlüssel A1”, daher halten wir eine Kopie im Speicher. Diese Kopie wird geladen, indem das gesamte Topic jedes Mal gelesen wird, wenn die Anwendung startet:

@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("Ereignis konnte nicht geschrieben werden: " + 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);
    }
}

Erklärung der einzelnen Teile:

send(TOPIC, event.orderId(), event) verwendet orderId als Schlüssel. Dadurch fallen alle Ereignisse einer Bestellung in die gleiche Partition und behalten die richtige Reihenfolge.
.get(...) wartet auf die Bestätigung durch den Broker. Ein Command gilt nur dann als erfolgreich, wenn das Ereignis im Log vorhanden ist.
Der Listener verwendet eine zufällige groupId. Bei jedem Start ist es eine neue Gruppe, kombiniert mit earliest, liest die gesamte Historie in die Map.

6.2. Command Service

@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("Bestellung nicht gefunden: " + orderId);
        }
        return Order.replay(history);
    }
}

Der Verarbeitungsfluss ist immer gleich: Load Historie → Replay → Command aufrufen → Ereignis append. Es gibt keine UPDATE-Anweisung.

Hinweis: Die Map im Speicher wird asynchron über den Listener aktualisiert. Wenn du addItem direkt nach createOrder aufrufst, hat die Map möglicherweise Created noch nicht empfangen. Außerdem hat Kafka keine optimistische Concurrency-Kontrolle wie “schreibe nur, wenn die aktuelle Version 4 ist”. Zwei Instanzen, die gleichzeitig eine Bestellung verarbeiten, können zwei widersprüchliche Ereignisse schreiben. In der Produktion solltest du alle Commands für eine orderId an einen einzigen Consumer weiterleiten (Command-Topic mit Schlüssel nach orderId), oder Kafka Streams mit State Store verwenden. Eine weitere Option ist, einen dedizierten Event Store oder eine Datenbank als primäre Schreibquelle zu verwenden und dann über Kafka zu publizieren.

7. Projektion: Read Model aufbauen

Das Event Log ist hervorragend zum Schreiben, aber nicht praktisch für Abfragen wie “liste alle bezahlten Bestellungen des Kunden C7”. Projektion löst dieses Problem. Es ist ein Consumer, der Ereignisse liest und ein für das Lesen optimiertes Modell verwaltet. Das ist der “Q” Teil in 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();
    }
}

Dieses Beispiel verwendet ConcurrentHashMap der Einfachheit halber. In der Praxis würdest du in PostgreSQL, Elasticsearch oder Redis schreiben. Der wichtige Punkt ist, dass die Projektion eine feste groupId hat, daher merkt sich Kafka den Offset und verarbeitet nur neue Ereignisse nach jedem Neustart.

Die Projektion muss idempotent sein. Kafka garantiert at-least-once, daher könnte ein Ereignis nach einem Rebalance erneut zugestellt werden. Mit einer Datenbank kannst du den zuletzt verarbeiteten Offset pro Partition speichern und ältere Datensätze überspringen.

7.1. Replay zum Neu-Erstellen des Read Model

Angenommen, die Projektion hat einen Bug, der den total falsch berechnet. Du behebst den Code, löschst die Read-Model-Daten, stoppst die Anwendung und setzt den Offset der Gruppe zurück:

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

Beim Neustart liest die Projektion die gesamte Historie und erstellt das korrekte Read Model. Um einen völlig neuen Report hinzuzufügen, schreibst du einfach einen neuen Listener mit einer neuen groupId. Dieser Report erhält automatisch alle Daten von Anfang an. Das ist der größte Vorteil des Speicherns von Ereignissen statt Status.

8. Häufige Fehler und wann man Event Sourcing nicht verwenden sollte

Änderung des Ereignis-Schemas. Alte Ereignisse existieren für immer. Füge nur neue Felder mit Standardwerten hinzu und lösche oder änderst nicht die Bedeutung alter Felder. Wenn eine große Änderung notwendig ist, erstelle einen neuen Ereignistyp (ItemAddedV2) oder schreibe einen “Upcaster”, der alte Ereignisse beim Lesen konvertiert. Bei großen Systemen solltest du Schema Registry mit Avro oder Protobuf verwenden.

Status in Events einbetten. Ein Ereignis wie OrderUpdated { gesamte Bestellung } ist nur aufgemotzte CRUD. Ereignisse sollten die geschäftliche Absicht beschreiben, wie ItemAdded oder OrderCancelled.

Zu langer Replay. Ein Aggregate mit Tausenden von Ereignissen wird langsam geladen. Die Lösung ist Snapshot: Speichere regelmäßig den Status zusammen mit version in einem compacted Topic, und beim Laden liest du zuerst den Snapshot und spielst dann nur die Ereignisse danach ab.

Side Effects in apply aufrufen. Wenn apply eine E-Mail versendet, wird bei jedem Replay eine E-Mail erneut versendet. apply darf nur den In-Memory-Status ändern.

Persönliche Daten und Löschungsrecht. Ein unbewegliches Log widerspricht der Anforderung, Nutzerdaten zu löschen. Eine übliche Technik ist Crypto-Shredding: Verschlüssele persönliche Daten mit einem eindeutigen Schlüssel pro Kunde, und beim Löschen wirfst du den Schlüssel weg.

Du solltest nicht Event Sourcing verwenden, wenn:

Die Domain hauptsächlich einfaches CRUD ist und niemand die Historie braucht.
Das Team mit eventueller Konsistenz zwischen Schreib- und Read-Seite nicht vertraut ist.
Du unmittelbar nach dem Schreiben über mehrere Aggregate hinweg konsistente Abfragen brauchst.

Hinweis: Kafka ist ein ausgezeichnetes Event Log, aber kein vollständiger Event Store. Es fehlen zwei Dinge: schnelles Lesen pro Aggregate und bedingtes Schreiben nach Version. Viele Systeme verwenden eine Datenbank oder einen dedizierten Event Store als Quelle der Wahrheit, dann Kafka (oft über Outbox Pattern) zur Verteilung von Ereignissen an Projektionen und andere Services.

9. Fazit

In diesem Artikel hast du Bestellungen als unveränderliche Abfolge von Ereignissen in Kafka gespeichert statt als einen Datensatz. Du hast Ereignisse mit sealed interface und record modelliert, Status mit Order.replay rekonstruiert und Geschäftsregeln vor dem Append überprüft. Du hast auch ein Read Model mit Projektion aufgebaut und es durch Zurücksetzen des Offsets neu erstellt.

Drei Punkte zum Merken:

Verwende orderId als Schlüssel, um die Reihenfolge von Ereignissen pro Bestellung zu bewahren.
Deaktiviere zeitbasierte Retention und verwende keine Compaction für Ereignis-Topics.
Seien dir der Concurrency-Grenzen von Kafka bewusst, bevor du es als einzige Quelle der Wahrheit siehst.

Bei richtiger Anwendung gibt dir Event Sourcing eine vollständige Historie und die Möglichkeit, neue Read Models ohne Migration zu erstellen.