Outbox-Muster mit Kafka: Keine verlorenen Events beim Datenbankschreiben

1. Überblick

Ein Service muss normalerweise zwei Dinge nacheinander tun: Daten in die Datenbank schreiben und Events an Kafka senden, damit andere Services diese empfangen. Diese beiden Systeme teilen sich keine Transaktion. Wenn ein Schritt erfolgreich ist und der andere fehlschlägt, verliert das System die Konsistenz. Das Ergebnis: Die Bestellung wurde gespeichert, aber niemand hat das Event OrderPlaced erhalten.

Dieser Artikel erklärt das Dual-Write-Problem und implementiert das Outbox-Muster mit Java 21, Spring Boot 3.x und Spring Kafka. Nach dem Lesen wirst du einen Message-Relay haben, der Events aus der Outbox-Tabelle an Kafka weiterleitet, inklusive Tests und idempotenten Consumer.

2. Das Dual-Write-Problem

Schaue dir die folgende “naive” Schreibweise an:

@Transactional
public Order placeOrder(String customerId, BigDecimal amount) {
    Order order = orderRepository.save(new Order(customerId, amount));
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
    return order;
}

Dieser Code hat mindestens zwei Fehlerszenarios:

• Kafka sendet erfolgreich, Transaktion wird zurückgerollt. Zum Beispiel ein Fehler beim Commit oder ein Constraint-Verstoß. Das “Phantom”-Event ist bereits rausgegangen für eine Bestellung, die nicht existiert.
• Transaktion committed, Kafka-Versand schlägt fehl. Der Broker könnte unerreichbar sein oder die Anwendung crasht direkt nach dem Commit. Die Bestellung ist in der Datenbank, aber das Event ist weg.

Wenn du die Reihenfolge der beiden Befehle umkehrst, löst das das Grundproblem nicht. Zwei Operationen auf zwei verschiedenen Systemen können nicht atomar sein, ohne verteilte Transaktionen zu verwenden. Die meisten Teams wollen XA/2PC auch nicht verwenden, da es komplex und langsam ist.

3. Idee des Outbox-Musters

Das Outbox-Muster verwandelt “in Datenbank schreiben + an Kafka senden” in “in Datenbank schreiben + in Datenbank schreiben”:

1. Innerhalb derselben Transaktion speichert der Service die Order und fügt eine Zeile in die Tabelle outbox_event ein.
2. Ein separater Prozess namens Message Relay liest die ungesendeten Zeilen aus der Outbox-Tabelle, sendet sie an Kafka und markiert sie als gesendet.

Da Schritt 1 nur eine Datenbank betrifft, ist er absolut atomar: Wenn es eine Bestellung gibt, gibt es garantiert ein Event. Schritt 2 kann so oft wiederholt werden, wie man möchte. Dafür erhältst du die Semantik at-least-once: Ein Event kann dupliziert gesendet werden, geht aber niemals verloren.

4. Implementierung mit Spring Boot

4.1. Abhängigkeiten

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Als Nächstes die sichere Producer-Konfiguration:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.properties.enable.idempotence=true

acks=all zwingt den Broker, erst zu bestätigen, nachdem alle Replicas geschrieben haben. enable.idempotence=true hilft dem Producer, keine Duplikate zu erzeugen, wenn er auf Netzwerk-Ebene erneut versucht.

4.2. Outbox-Tabelle und Entity

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP    NOT NULL,
    published_at   TIMESTAMP
);
CREATE INDEX idx_outbox_unpublished ON outbox_event (created_at) WHERE published_at IS NULL;

Der partielle Index enthält nur ungesendete Zeilen. Dadurch bleibt die Abfrage des Relays schnell, selbst wenn die Tabelle bereits Millionen von alten Zeilen enthält.

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    private UUID id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    @Column(columnDefinition = "text")
    private String payload;
    private Instant createdAt;
    private Instant publishedAt;

    protected OutboxEvent() {
    }

    public OutboxEvent(String aggregateType, String aggregateId, String eventType, String payload) {
        this.id = UUID.randomUUID();
        this.aggregateType = aggregateType;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = Instant.now();
    }

    public void markPublished(Instant at) {
        this.publishedAt = at;
    }

    public UUID getId() { return id; }
    public String getAggregateId() { return aggregateId; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public Instant getPublishedAt() { return publishedAt; }
}

Die id dient auch als eindeutige Event-ID. Der Consumer wird sie verwenden, um Duplikate zu filtern (siehe Abschnitt 6). aggregateId wird der Schlüssel der Kafka-Nachricht.

4.3. Order und Event in derselben Transaktion speichern

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

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public OrderService(OrderRepository orderRepository,
                        OutboxEventRepository outboxRepository,
                        ObjectMapper objectMapper) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public Order placeOrder(String customerId, BigDecimal amount) {
        Order order = orderRepository.save(new Order(customerId, amount));

        var event = new OrderPlacedEvent(order.getId(), customerId, amount);
        outboxRepository.save(new OutboxEvent(
                "Order", order.getId().toString(), "OrderPlaced", toJson(event)));

        return order;
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("Ereignis konnte nicht serialisiert werden", e);
        }
    }
}

Der kritische Punkt liegt bei @Transactional. Beide save-Operationen committen oder rollback zusammen. OrderService weiß nicht mehr über Kafka, sodass kein Netzwerk-Aufruf innerhalb der Transaktion stattfindet.

4.4. Message Relay, das Events an Kafka weiterleitet

Zunächst schreibe ein Repository mit gesperrter Abfrage:

public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    @Query(value = """
            SELECT * FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT :batchSize
            FOR UPDATE SKIP LOCKED
            """, nativeQuery = true)
    List<OutboxEvent> findBatchToPublish(@Param("batchSize") int batchSize);
}

FOR UPDATE SKIP LOCKED (PostgreSQL, MySQL 8+) ermöglicht mehrere Instanzen der Anwendung parallel. Jede Instance sperrt einen eigenen Batch und überspringt Zeilen, die andere Instanzen gerade verarbeiten. So sendet keine Instanz dasselbe Event gleichzeitig.

Als Nächstes das Relay:

@Component
public class OutboxRelay {

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

    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(OutboxEventRepository outboxRepository,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPending() {
        for (OutboxEvent event : outboxRepository.findBatchToPublish(100)) {
            try {
                kafkaTemplate.send(toRecord(event)).get(5, TimeUnit.SECONDS);
                event.markPublished(Instant.now());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                log.warn("Versand von Event {} fehlgeschlagen, wird später erneut versucht", event.getId(), e);
                break; // Stopp, um die Reihenfolge zu bewahren; beim nächsten Durchlauf erneut versuchen
            }
        }
    }

    private ProducerRecord<String, String> toRecord(OutboxEvent event) {
        var record = new ProducerRecord<>("orders", event.getAggregateId(), event.getPayload());
        record.headers()
              .add("eventId", event.getId().toString().getBytes(StandardCharsets.UTF_8))
              .add("eventType", event.getEventType().getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

Schritt für Schritt:

• @Scheduled(fixedDelay = 1000) läuft das Relay eine Sekunde nach Ende des letzten Durchlaufs. Vergiss nicht, @EnableScheduling in eine Konfigurationsklasse zu schreiben.
• .get(5, TimeUnit.SECONDS) wartet auf die Bestätigung des Brokers. Das Relay markiert publishedAt nur, wenn sicher ist, dass Kafka die Nachricht erhalten hat.
• markPublished ändert die verwaltete Entity. JPA generiert automatisch ein UPDATE-Statement beim Commit der Transaktion; du brauchst save nicht aufzurufen.
• Bei einem Fehler macht das Relay einen break statt die Zeile zu überspringen. Events derselben Bestellung gehen so nicht durcheinander.
• Der Key ist aggregateId, daher fallen alle Events einer Order in dieselbe Partition und bewahren ihre Reihenfolge in Kafka.

Das Log beim Broker-Ausfall sieht etwa so aus:

WARN  OutboxRelay : Versand von Event 3f6c1e0a-... fehlgeschlagen, wird später erneut versucht
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

Wenn der Broker wieder aktiv ist, wird beim nächsten Durchlauf vom richtigen Event an weitermacht.

Hinweis: Falls die Anwendung crasht, nachdem Kafka die Nachricht empfangen hat, aber bevor die Transaktion committed, bleibt die Outbox-Zeile “ungesendet”. Der nächste Durchlauf sendet erneut. Das ist genau at-least-once, und der Consumer muss Duplikate verarbeiten können.

5. Testen

Der erste Test prüft, dass jede Bestellung genau eine Outbox-Zeile erzeugt:

@DataJpaTest
@Import({OrderService.class, JacksonAutoConfiguration.class})
class OrderServiceTest {

    @Autowired
    OrderService orderService;

    @Autowired
    OutboxEventRepository outboxRepository;

    @Test
    void placeOrder_shouldWriteOutboxEventInSameTransaction() {
        Order order = orderService.placeOrder("customer-42", new BigDecimal("150000"));

        List<OutboxEvent> events = outboxRepository.findAll();

        assertThat(events).hasSize(1);
        OutboxEvent event = events.getFirst();
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.getEventType()).isEqualTo("OrderPlaced");
        assertThat(event.getPayload()).contains("\"customerId\":\"customer-42\"");
        assertThat(event.getPublishedAt()).isNull();
    }
}

Native Queries werden beim Start nicht validiert. Deshalb funktioniert H2 trotzdem für diesen Test, auch wenn es SKIP LOCKED nicht vollständig unterstützt. Um das Relay mit echter Query zu testen, solltest du PostgreSQL über Testcontainers verwenden.

Der zweite Test stellt sicher, dass das Relay nicht den Status zu gesendet setzt, wenn Kafka fehlschlägt:

class OutboxRelayTest {

    OutboxEventRepository repository = mock(OutboxEventRepository.class);
    KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
    OutboxRelay relay = new OutboxRelay(repository, kafkaTemplate);

    @Test
    void shouldKeepEventUnpublished_whenKafkaFails() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker down")));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNull();
    }

    @Test
    void shouldMarkPublished_whenKafkaAcknowledges() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.completedFuture(null));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNotNull();
    }
}

Diese beiden Tests fixieren das Kernverhalten: Markiere als gesendet nur, wenn der Broker bestätigt hat.

6. Konsumenten-Seite: Idempotente Verarbeitung

Da Duplikate möglich sind, muss der Consumer die bereits verarbeiteten eventIds merken. Diese Merkung muss in derselben Transaktion wie die Geschäftslogik stattfinden:

@Component
public class InventoryListener {

    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public InventoryListener(ProcessedEventRepository processedEvents,
                             InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    @Transactional
    public void onOrderPlaced(@Header("eventId") String eventId, String payload) {
        if (processedEvents.existsById(eventId)) {
            return; // bereits verarbeitet, überspringe das Duplikat
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

Die Tabelle processed_event sollte event_id als Primary Key haben. Falls zwei Threads gleichzeitig ein Duplikat verarbeiten, wird einer durch die Constraint blockiert.

7. Häufige Fehler und wann man das Muster nicht verwenden sollte

• Kafka aus OrderService aufrufen “zum Schneller werden”. Das ist wieder Dual Write. Alle Events müssen über die Outbox-Tabelle gehen.
• Vergessen, die Outbox-Tabelle zu räumen. Die Tabelle wird schnell riesig. Habe einen Job, der Zeilen älter als einige Tage mit published_at löscht.
• Fehlerhafte Zeilen überspringen und weitermachen. Das bricht die Event-Reihenfolge. Falls ein Payload dauerhaft fehlerhaft ist, verschiebe ihn in einen Status FAILED zur Untersuchung; retry nicht endlos.
• Polling zu häufig. Ein 100-ms-Zyklus mit mehreren Instanzen erzeugt erhebliche Last auf der Datenbank. Ein Zyklus von 0,5–2 Sekunden ist meist ausreichend.
• Sehr niedrige Latenz oder sehr großen Durchsatz nötig. Verwende dann CDC (Change Data Capture) statt Polling. Beispiel: Debezium liest das WAL/binlog und pusht Outbox-Zeilen direkt an Kafka. Die Outbox-Tabelle bleibt, nur das Relay ändert sich.

Hinweis: Verwende nicht das Outbox-Muster, falls Events nicht kritisch sind, z. B. Metriken oder Benutzerverhalten-Logs können verlorengehen. Dann ist direktes Senden an Kafka viel einfacher. Umgekehrt: Kafka Transactions (transactional.id) ersetzen nicht das Outbox-Muster; sie garantieren Atomarität nur zwischen Kafka-Topics, nicht zwischen deiner Datenbank und Kafka.

8. Fazit

Dual Write zwischen Datenbank und Kafka hat immer Lücken, die Events verlieren oder duplizieren. Das Outbox-Muster schließt diese Lücke in drei Schritten. Erstens: Schreibe Events in die Outbox-Tabelle innerhalb derselben Transaktion wie die Geschäftsdaten. Zweitens: Nutze ein separates Relay, um Events an Kafka zu pushen und markiere sie als gesendet nur, wenn der Broker bestätigt hat. Drittens: Verarbeite Duplikate auf der Consumer-Seite idempotent, um die at-least-once-Semantik zu handhaben.

Du hast jetzt alle Teile: Schema mit partiellen Index, OrderService ohne Kafka-Abhängigkeit, Relay mit FOR UPDATE SKIP LOCKED für mehrere Instanzen, Tests für das Kernverhalten und ein duplikat-resistentes Consumer. Wenn dein System wächst, kannst du Polling durch CDC ersetzen, ohne Business-Code zu ändern.