Change Data Capture mit Debezium und Kafka

1. Überblick

Wenn ein Service wissen muss, dass sich Daten in einer anderen Datenbank gerade geändert haben, ist die übliche Methode, in die Datenbank zu schreiben und zusätzlich eine Nachricht zu Kafka zu senden. Dies wird Dual-Write genannt und führt leicht zu Datenlücken. Change Data Capture (CDC) löst dieses Problem, indem es direkt aus dem Transaktionslog der Datenbank liest.

In diesem Artikel wirst du PostgreSQL, Kafka und Debezium mit Docker Compose aufbauen und anschließend einen Connector registrieren. Danach schreibst du einen Spring Boot Consumer, um die INSERT-, UPDATE- und DELETE-Events der Tabelle orders zu verarbeiten.

2. Warum CDC nötig ist

Sieh dir den folgenden bekannten Dual-Write-Code an:

@Transactional
public void placeOrder(Order order) {
    orderRepository.save(order);
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
}

Dieser Code hat zwei Schwachstellen:

• Wenn Kafka fehlschlägt, nachdem save erfolgreich ist, hat die Datenbank eine Bestellung, aber kein Event wurde gesendet.
• Wenn send erfolgreich ist, aber die Transaktion zurückgerollt wird, erhält der Consumer ein Event für eine Bestellung, die nicht existiert.

CDC kehrt diesen Ablauf um. Die Anwendung schreibt nur in die Datenbank. Ein separater Prozess liest das Transaktionslog (WAL bei PostgreSQL, binlog bei MySQL) und gibt für jede committed Änderung ein Event aus. Da das Log nur committed Änderungen enthält, stimmen Events immer mit den echten Daten überein.

Debezium ist eine Sammlung von CDC-Connectoren, die auf Kafka Connect laufen. Der Datenfluss sieht folgendermaßen aus:

Anwendung ──schreibt──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Kafka Topic ──> Consumer

3. Umgebung aufbauen

3.1. Docker Compose

Du benötigst drei Container: PostgreSQL mit aktivierter logical replication, Kafka im KRaft-Modus und Kafka Connect mit dem vorinstallierten Debezium-Plugin.

services:
  postgres:
    image: postgres:16
    command: ["postgres", "-c", "wal_level=logical"]
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: shop
    ports:
      - "5432:5432"

  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  connect:
    image: quay.io/debezium/connect:2.7
    depends_on: [kafka, postgres]
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses

Es gibt drei wichtige Punkte in dieser Datei:

• wal_level=logical ist erforderlich. Ohne diesen Parameter kann Debezium keinen Replication Slot erstellen.
• Kafka hat zwei Listener. Der Container nutzt kafka:29092, während deine Spring Boot Anwendung auf deinem Rechner localhost:9092 verwendet.
• Kafka Connect speichert Konfiguration, Offset und Status in drei separaten Topics. Dadurch kann der Connector nach einem Neustart dort weitermachen, wo er aufgehört hat.

Starten und erstelle die Tabelle:

docker compose up -d
docker compose exec postgres psql -U postgres -d shop -c "
  CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    customer_email VARCHAR(255) NOT NULL,
    status VARCHAR(32) NOT NULL,
    total NUMERIC(10,2) NOT NULL
  );
  ALTER TABLE orders REPLICA IDENTITY FULL;"

REPLICA IDENTITY FULL fordert PostgreSQL auf, die gesamte alte Zeile ins WAL zu schreiben. Ohne diesen Befehl ist das Feld before in UPDATE-Events null, und DELETE-Events enthalten nur den Primärschlüssel.

3.2. Connector registrieren

Kafka Connect stellt eine REST-API auf Port 8083 bereit. Du registrierst den PostgreSQL-Connector wie folgt:

curl -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "table.include.list": "public.orders",
    "slot.name": "shop_slot",
    "publication.autocreate.mode": "filtered",
    "decimal.handling.mode": "string",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}'

Bedeutung der wichtigen Eigenschaften:

• plugin.name: pgoutput verwendet das in PostgreSQL 10+ eingebaute Decoding-Plugin, sodass du keine Extension installieren musst.
• topic.prefix kombiniert mit Schema- und Tabellenname, um den Topic-Namen zu erstellen. Hier ist das Topic shop.public.orders.
• table.include.list begrenzt, welche Tabellen erfasst werden.
• decimal.handling.mode: string sendet NUMERIC-Spalten als Zeichenkettenformat "149.90". Standardmäßig codiert Debezium diesen Wert als Base64-Bytes, was sehr schwer zu lesen ist.
• schemas.enable: false entfernt den schema-Teil aus jeder JSON-Nachricht, was die Payload kompakter macht.

Überprüfe den Connector-Status:

curl -s localhost:8083/connectors/shop-connector/status

{"name":"shop-connector","connector":{"state":"RUNNING","worker_id":"172.18.0.4:8083"},
 "tasks":[{"id":0,"state":"RUNNING","worker_id":"172.18.0.4:8083"}],"type":"source"}

4. Struktur von Change Events

Füge eine Bestellung hinzu und aktualisiere sie dann:

docker compose exec postgres psql -U postgres -d shop -c "
  INSERT INTO orders (customer_email, status, total) VALUES ('an@example.com', 'NEW', 149.90);
  UPDATE orders SET status = 'PAID' WHERE id = 1;"

Lese dann das Topic mit dem in Kafkas Image enthaltenen Console Consumer:

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic shop.public.orders --from-beginning

Das UPDATE-Event sieht folgendermaßen aus (mit verkürztem source-Feld):

{
  "before": {"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
  "after":  {"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
  "source": {"connector":"postgresql","db":"shop","table":"orders","lsn":24023128, ...},
  "op": "u",
  "ts_ms": 1760170000123
}

Das Feld op zeigt die Art der Änderung an:

op	Bedeutung	before	after
c	INSERT	null	neue Zeile
u	UPDATE	alte Zeile	neue Zeile
d	DELETE	alte Zeile	null
r	Lesen im anfänglichen Snapshot	null	vorhandene Zeile

Wenn du eine Zeile löschst, sendet Debezium zusätzlich einen Tombstone, d.h. eine Nachricht mit demselben Key, aber Wert null. Der Tombstone ermöglicht es Kafkas Log Compaction, diesen Key völlig zu bereinigen. Dein Consumer muss mit einem null-Wert umgehen können.

5. Events mit Spring Boot konsumieren

5.1. Abhängigkeiten und Konfiguration

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

Die Versionen der obigen Abhängigkeiten werden von spring-boot-starter-parent 3.x verwaltet.

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-sync
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

Hier empfängt der Consumer den Wert als String und parst ihn selbst. Dieser Ansatz gibt dir mehr Kontrolle über Tombstones und JSON-Fehler als die Verwendung von JsonDeserializer.

5.2. Modell und Parser

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderRow(
        Long id,
        @JsonProperty("customer_email") String customerEmail,
        String status,
        BigDecimal total) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderChangeEvent(
        OrderRow before,
        OrderRow after,
        String op,
        @JsonProperty("ts_ms") Long tsMs) {
}

@JsonIgnoreProperties(ignoreUnknown = true) ignoriert das source-Feld und alle neuen Felder, die Debezium später hinzufügen könnte. Jackson konvertiert automatisch den String "149.90" zu BigDecimal, sodass du keine Präzision verlierst wie bei der Verwendung von double.

@Component
public class OrderChangeParser {

    private final ObjectMapper objectMapper;

    public OrderChangeParser(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    public Optional<OrderChangeEvent> parse(String value) {
        if (value == null) {
            return Optional.empty(); // tombstone
        }
        try {
            return Optional.of(objectMapper.readValue(value, OrderChangeEvent.class));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Invalid Debezium event", e);
        }
    }
}

Der Parser gibt Optional.empty() für Tombstones zurück, sodass der Listener diesen Fall einfach überspringen kann.

5.3. Listener

@Component
public class OrderCdcListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCdcListener.class);
    private final OrderChangeParser parser;

    public OrderCdcListener(OrderChangeParser parser) {
        this.parser = parser;
    }

    @KafkaListener(topics = "shop.public.orders")
    public void onOrderChange(@Payload(required = false) String value) {
        parser.parse(value).ifPresent(event -> {
            switch (event.op()) {
                case "c", "r" -> log.info("Order created: {}", event.after());
                case "u" -> log.info("Order {} status {} -> {}",
                        event.after().id(), event.before().status(), event.after().status());
                case "d" -> log.info("Order deleted: {}", event.before().id());
                default -> log.warn("Unknown op: {}", event.op());
            }
        });
    }
}

@Payload(required = false) ermöglicht es Spring, null an die Methode zu übergeben, wenn ein Tombstone auftritt. Ohne diese Annotation würde Spring einen Konvertierungsfehler melden. Der Branch "r" wird mit "c" zusammengefasst, da für einen Consumer eine vorhandene Zeile im Snapshot einer gerade erstellten Zeile ähnelt.

Wenn du die Anwendung mit den Daten aus Abschnitt 4 ausführst, siehst du:

INFO  OrderCdcListener : Order created: OrderRow[id=1, customerEmail=an@example.com, status=NEW, total=149.90]
INFO  OrderCdcListener : Order 1 status NEW -> PAID

6. Parser testen

Die Parse-Logik ist der fehleranfälligste Teil, wenn sich Debezium oder das Schema ändert. Du kannst diesen Teil testen, ohne Kafka zu benötigen:

class OrderChangeParserTest {

    private final OrderChangeParser parser = new OrderChangeParser(new ObjectMapper());

    @Test
    void parsesUpdateEvent() {
        String json = """
            {"before":{"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
             "after":{"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
             "source":{"connector":"postgresql"},
             "op":"u","ts_ms":1760170000123}
            """;

        OrderChangeEvent event = parser.parse(json).orElseThrow();

        assertThat(event.op()).isEqualTo("u");
        assertThat(event.before().status()).isEqualTo("NEW");
        assertThat(event.after().status()).isEqualTo("PAID");
        assertThat(event.after().total()).isEqualByComparingTo("149.90");
    }

    @Test
    void returnsEmptyForTombstone() {
        assertThat(parser.parse(null)).isEmpty();
    }

    @Test
    void rejectsMalformedJson() {
        assertThatThrownBy(() -> parser.parse("{not json"))
                .isInstanceOf(IllegalArgumentException.class);
    }
}

Der erste Test überprüft die Zuordnung des Spaltennamens customer_email und des Typs BigDecimal. Der zweite Test stellt sicher, dass Tombstones nicht zum Fehler im Listener führen. Der dritte Test bestätigt, dass fehlerhaftes JSON ordnungsgemäß gemeldet wird, anstatt stillschweigend ignoriert zu werden.

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

Hinweis: Der Replication Slot speichert das WAL, bis Debezium bestätigt hat, dass es gelesen wurde. Wenn der Connector viele Tage lang gestoppt ist, wird das WAL zunehmen und könnte die Festplatte der Datenbank füllen. Du solltest pg_replication_slots überwachen und den Slot löschen, wenn du den Connector entfernst: SELECT pg_drop_replication_slot('shop_slot');.

Einige andere Fehler, auf die du stoßen wirst:

• Consumer ist nicht idempotent. Debezium und Kafka Connect garantieren At-Least-Once. Nach einem Neustart kann dasselbe Event zweimal ankommen. Verwende id zusammen mit source.lsn, um Duplikate zu überspringen, oder gestalte die Operation idempotent (z.B. Upsert).
• NUMERIC-Wert wird als unverständliches Base64 gesendet. Du hast decimal.handling.mode nicht konfiguriert. Ähnlich werden TIMESTAMP-Spalten standardmäßig als Anzahl der Mikrosekunden seit dem Epoch gesendet, also musst du diese manuell konvertieren.
• before ist immer null. Die Tabelle hat kein REPLICA IDENTITY FULL gesetzt.
• Anfänglicher Snapshot dauert zu lange. Bei großen Tabellen wird snapshot.mode=initial die gesamte Tabelle lesen, bevor es streamt. Erwäge no_data, wenn du nur Änderungen von jetzt an brauchst, oder verwende incremental snapshots.
• Datenleck von internen Daten. Das Event spiegelt die genaue Tabellenstruktur wider. Wenn du es direkt an einen anderen Service weitergibst, offenbarst du dein Datenbankschema. Für Business Events solltest du Debezium mit dem Outbox-Muster kombinieren: Die Anwendung schreibt in eine outbox-Tabelle innerhalb derselben Transaktion, und Debezium erfasst nur diese Tabelle.
• Production postgres-Konto verwenden. Erstelle einen eigenen Benutzer mit REPLICATION-Berechtigung und SELECT-Berechtigung auf den Tabellen, die erfasst werden sollen.

Du solltest CDC nicht verwenden, wenn:

• Du nur eine Anwendung hast und sie kann Events selbst in demselben Prozess generieren. Dann ist ein transaktionales Outbox mit einem einfachen Poller ausreichend.
• Du Events benötigst, die geschäftliche Bedeutung wie „OrderPaid” tragen, anstatt „Zeile X Spalte status geändert”. CDC zeigt nur, dass sich Daten ändern, nicht warum.
• Dein Betriebsteam noch nicht bereit ist, Kafka Connect, Replication Slots und Schemaentwicklung zu verwalten.

8. Fazit

In diesem Artikel hast du PostgreSQL mit logical replication, Kafka und Debezium aufgebaut und anschließend einen Connector zur Erfassung der orders-Tabelle registriert. Du hast auch die Struktur von Change Events mit den Feldern before, after und op untersucht und dann einen Spring Boot Consumer geschrieben, der Tombstones verarbeitet, zusammen mit Tests für die Parse-Logik.

CDC beseitigt das Risiko von Dual-Writes, da die Datenbank zur einzigen Quelle der Wahrheit wird. Im Gegenzug musst du Replication Slots überwachen, idempotente Consumer schreiben und das Outbox-Muster in Betracht ziehen, wenn du klare Business Events benötigst.