{"id":681,"date":"2026-06-06T00:33:41","date_gmt":"2026-06-05T23:33:41","guid":{"rendered":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/?p=681"},"modified":"2026-06-17T00:35:14","modified_gmt":"2026-06-16T23:35:14","slug":"debezium-change-data-capture-mit-java-und-kafka-connect","status":"publish","type":"post","link":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/?p=681","title":{"rendered":"Debezium \u2014 Change Data Capture mit Java und Kafka Connect"},"content":{"rendered":"\n

In Microservice-Architekturen m\u00fcssen Daten\u00e4nderungen oft an mehrere Dienste propagiert werden \u2014 ein neuer Kunde, eine aktualisierte Bestellung, eine stornierte Buchung. Anstatt jeden Dienst aktiv die Datenbank pollen zu lassen, nutzt Change Data Capture (CDC)<\/strong> die Write-Ahead-Logs der Datenbank selbst als Event-Quelle. Debezium<\/strong> ist die f\u00fchrende Open-Source-CDC-Plattform, die Datenbanken wie PostgreSQL, MySQL, MongoDB und SQL Server in Echtzeit an Apache Kafka streamt.<\/p>\n\n\n\n

Das CDC-Prinzip<\/h2>\n\n\n\n

Traditionelle Synchronisationsans\u00e4tze pollen in regelm\u00e4\u00dfigen Abst\u00e4nden die Datenbank:<\/p>\n\n\n

Dienst<\/span> A<\/span> --<\/span>> [Poll: SELECT * FROM bestellungen WHERE updated > ?]<\/span> --<\/span>> Datenbank<\/span>\n<\/code><\/span>Code-Sprache:<\/span> CSS<\/span> (<\/span>css<\/span>)<\/span><\/small><\/pre>\n\n\n
Das belastet die Datenbank und liefert \u00c4nderungen nur mit Verz\u00f6gerung. Debezium hingegen liest das Write-Ahead-Log (WAL)<\/strong> der Datenbank:<\/p>\n\n\n
PostgreSQL WAL --> Debezium Connector --> Kafka Topic --> Consumer 1, 2, 3\n<\/code><\/span><\/pre>\n\n\n
Jede INSERT-, UPDATE- und DELETE-Operation wird sofort als Kafka-Event publiziert \u2014 ohne die Source-Datenbank zus\u00e4tzlich zu belasten.<\/p>\n\n\n\n
Architektur<\/h2>\n\n\n\n
Debezium setzt auf Kafka Connect<\/strong> als Laufzeitumgebung:<\/p>\n\n\n
\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510     \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510     \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n\u2502PostgreSQL\u2502\u2500\u2500\u2500\u2500\u25b6\u2502Debezium Connector\u2502\u2500\u2500\u2500\u2500\u25b6\u2502  Kafka   \u2502\n\u2502  (WAL)   \u2502     \u2502  (Kafka Connect) \u2502     \u2502  Topic   \u2502\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518     \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518     \u2514\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2518\n                                              \u2502\n                         \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u253c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n                         \u25bc                    \u25bc             \u25bc\n                   Suchindex-         Analyse-         Audit-\n                   Aktualisierung     Pipeline         Service\n<\/code><\/span><\/pre>\n\n\n
PostgreSQL-Connector konfigurieren<\/h2>\n\n\n\n
Der Debezium-Connector wird als JSON-Konfiguration an Kafka Connect \u00fcbergeben:<\/p>\n\n\n
{\n  \"name\"<\/span>: \"bestellungen-connector\"<\/span>,\n  \"config\"<\/span>: {\n    \"connector.class\"<\/span>: \"io.debezium.connector.postgresql.PostgresConnector\"<\/span>,\n    \"plugin.name\"<\/span>: \"pgoutput\"<\/span>,\n    \"database.hostname\"<\/span>: \"localhost\"<\/span>,\n    \"database.port\"<\/span>: \"5432\"<\/span>,\n    \"database.user\"<\/span>: \"debezium\"<\/span>,\n    \"database.password\"<\/span>: \"geheim\"<\/span>,\n    \"database.dbname\"<\/span>: \"shop\"<\/span>,\n    \"topic.prefix\"<\/span>: \"shop_cdc\"<\/span>,\n    \"table.include.list\"<\/span>: \"public.bestellungen\"<\/span>,\n    \"publication.autocreate.mode\"<\/span>: \"filtered\"<\/span>\n  }\n}\n<\/code><\/span>Code-Sprache:<\/span> JSON \/ JSON mit Kommentaren<\/span> (<\/span>json<\/span>)<\/span><\/small><\/pre>\n\n\n
PostgreSQL ben\u00f6tigt daf\u00fcr wal_level = logical<\/code> in der postgresql.conf<\/code>.<\/p>\n\n\n\n
Events verarbeiten in Java<\/h2>\n\n\n\n
Jeder Datenbank-Change wird als Kafka-Event mit einer spezifischen Struktur publiziert:<\/p>\n\n\n
{\n  \"before\"<\/span>: null<\/span>,\n  \"after\"<\/span>: {\n    \"id\"<\/span>: 42<\/span>,\n    \"kunde_id\"<\/span>: 100<\/span>,\n    \"betrag\"<\/span>: 99.90<\/span>,\n    \"status\"<\/span>: \"NEU\"<\/span>\n  },\n  \"op\"<\/span>: \"c\"<\/span>,\n  \"ts_ms\"<\/span>: 1718123456000<\/span>\n}\n<\/code><\/span>Code-Sprache:<\/span> JSON \/ JSON mit Kommentaren<\/span> (<\/span>json<\/span>)<\/span><\/small><\/pre>\n\n\n
Der op<\/code>-Wert steht f\u00fcr c<\/code> (create), u<\/code> (update), d<\/code> (delete) oder r<\/code> (snapshot). In Java verarbeitet man die Events mit einem Kafka-Streams-Topologie oder einem einfachen Consumer:<\/p>\n\n\n
import<\/span> org.apache.kafka.clients.consumer.KafkaConsumer;\nimport<\/span> org.apache.kafka.clients.consumer.ConsumerRecords;\n\nimport<\/span> java.time.Duration;\nimport<\/span> java.util.List;\nimport<\/span> java.util.Properties;\n\npublic class<\/span> BestellungsConsumer<\/span> <\/span>{\n\n    public static<\/span> void<\/span> main(String<\/span>[] args) {\n        var<\/span> props = new<\/span> Properties();\n        props.put(\"bootstrap.servers\"<\/span>, \"localhost:9092\"<\/span>);\n        props.put(\"group.id\"<\/span>, \"bestellungs-service\"<\/span>);\n        props.put(\"key.deserializer\"<\/span>,\n            \"org.apache.kafka.common.serialization.StringDeserializer\"<\/span>);\n        props.put(\"value.deserializer\"<\/span>,\n            \"org.apache.kafka.common.serialization.StringDeserializer\"<\/span>);\n\n        try<\/span> (var<\/span> consumer = new<\/span> KafkaConsumer<String<\/span>, String<\/span>>(props)) {\n            consumer.subscribe(List.of(\"shop_cdc.public.bestellungen\"<\/span>));\n\n            while<\/span> (true<\/span>) {\n                ConsumerRecords<String<\/span>, String<\/span>> records =\n                    consumer.poll(Duration.ofMillis(100<\/span>));\n                records.forEach(record -> {\n                    System.out.printf(\"Topic: %s, Key: %s, Value: %s%n\"<\/span>,\n                        record.topic(), record.key(), record.value());\n                });\n                consumer.commitSync();\n            }\n        }\n    }\n}\n<\/code><\/span>Code-Sprache:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
Das Outbox-Pattern<\/h2>\n\n\n\n
Ein verbreitetes Muster in Kombination mit CDC ist das Outbox-Pattern<\/strong>: Statt eine Nachricht direkt in Kafka zu publizieren, schreibt der Service ein Event in eine Outbox-Tabelle \u2014 innerhalb derselben Datenbank-Transaktion. Debezium streamt die Outbox-Tabelle dann nach Kafka:<\/p>\n\n\n
<em>-- Transaktionale Outbox-Tabelle<\/em>\nCREATE TABLE outbox (\n    id UUID PRIMARY KEY,\n    aggregatetype VARCHAR(255<\/span>) NOT NULL<\/span>,\n    aggregateid VARCHAR(255<\/span>) NOT NULL<\/span>,\n    type VARCHAR(255<\/span>) NOT NULL<\/span>,\n    payload JSONB NOT NULL<\/span>,\n    timestamp TIMESTAMP DEFAULT<\/span> NOW()\n);\n<\/code><\/span>Code-Sprache:<\/span> PHP<\/span> (<\/span>php<\/span>)<\/span><\/small><\/pre>\n\n
<em>\/\/ Innerhalb einer @Transactional-Methode<\/em><\/span>\nbestellungsRepo.save(bestellung);\noutboxRepo.save(new<\/span> OutboxEvent(\n    UUID.randomUUID(),\n    \"Bestellung\"<\/span>,\n    bestellung.getId().toString(),\n    \"BestellungErstellt\"<\/span>,\n    toJson(bestellung)\n));\n\n<em<\/span>><\/span>\/\/ Debezium erwartet den Outbox Event Router SMT:<\/em<\/span>><\/span><\/span>\n<em<\/span>><\/span>\/\/ transforms=outbox<\/em<\/span>><\/span><\/span>\n<em<\/span>><\/span>\/\/ transforms.outbox.type=io.debezium.transforms.outbox.EventRouter<\/em<\/span>><\/span><\/span>\n<\/code><\/span>Code-Sprache:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
So ist garantiert, dass die Bestellung und das dazugeh\u00f6rige Event atomar persistiert werden \u2014 ohne verteilte Transaktionen (2PC).<\/p>\n\n\n\n
Embedded Debezium Engine<\/h2>\n\n\n\n
F\u00fcr Anwendungen, die kein vollst\u00e4ndiges Kafka-Cluster betreiben wollen, bietet Debezium eine Embedded Engine<\/strong>:<\/p>\n\n\n
var<\/span> engine = DebeziumEngine.create(Connect.class)\n    .using(props)\n    .notifying(record -> {\n        <em>\/\/ Event-Verarbeitung direkt in der JVM<\/em><\/span>\n        System.out.println(\"Change: \"<\/span> + record.key() + \" -> \"<\/span> + record.value());\n    })\n    .build();\n\nExecutorService executor = Executors.newSingleThreadExecutor();\nexecutor.execute(engine);\n<\/code><\/span>Code-Sprache:<\/span> JavaScript<\/span> (<\/span>javascript<\/span>)<\/span><\/small><\/pre>\n\n\n
Die Embedded Engine l\u00e4uft im selben Java-Prozess und verzichtet auf Kafka Connect \u2014 ideal f\u00fcr Tests und Single-Service-Szenarien.<\/p>\n\n\n\n
Fazit<\/h2>\n\n\n\n
Debezium bringt CDC in die Java-Welt und entkoppelt Daten\u00e4nderungen von ihrer Verarbeitung. In Kombination mit dem Outbox-Pattern entstehen zuverl\u00e4ssige, transaktionssichere Event-getriebene Architekturen. F\u00fcr Java-Entwickler, die mit Microservices und asynchroner Kommunikation arbeiten, ist Debezium ein unverzichtbares Werkzeug im Werkzeugkasten.<\/p>\n","protected":false},"excerpt":{"rendered":"
In Microservice-Architekturen m\u00fcssen Daten\u00e4nderungen oft an mehrere Dienste propagiert werden \u2014 ein neuer Kunde, eine aktualisierte Bestellung, eine stornierte Buchung. Anstatt jeden Dienst aktiv die Datenbank pollen zu lassen, nutzt Change Data Capture (CDC) die Write-Ahead-Logs der Datenbank selbst als Event-Quelle. Debezium ist die f\u00fchrende Open-Source-CDC-Plattform, die Datenbanken wie PostgreSQL, MySQL, MongoDB und SQL Server in Echtzeit an Apache […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4],"tags":[],"class_list":["post-681","post","type-post","status-publish","format-standard","hentry","category-plain_java"],"_links":{"self":[{"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=\/wp\/v2\/posts\/681","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=681"}],"version-history":[{"count":1,"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=\/wp\/v2\/posts\/681\/revisions"}],"predecessor-version":[{"id":682,"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=\/wp\/v2\/posts\/681\/revisions\/682"}],"wp:attachment":[{"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=681"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=681"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.xn--javaeinfacherklrt-4qb.de\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=681"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}