Change Data Capture (CDC): Architecture, Tools, and Kafka Integration

Change Data Capture (CDC) is a software design pattern that identifies and tracks data changes (inserts, updates, and deletes) in a source database and delivers them in real time to downstream destinations, such as Apache Kafka, data lakes, or search indices.

Why CDC Outperforms Traditional Batch Polling

• Zero Database Polling Overhead: Traditional polling requires frequent SELECT queries checking timestamp columns, which taxes database CPU. CDC directly tails transaction logs (e.g., Oracle Redo Logs, PostgreSQL WAL, MySQL Binlog).
• Complete Audit Trail (Captures Deletes): Polling cannot easily detect deleted rows without soft-delete columns. Log-based CDC captures every committed DML operation including hard deletes.
• Sub-Second Latency: Events stream as soon as transactions commit, enabling real-time analytics and event-driven microservices.

CDC Tool Ecosystem: Open Source vs. Enterprise

• Debezium (Open Source): Built on Apache Kafka Connect, Debezium provides connectors for PostgreSQL, MySQL, SQL Server, MongoDB, and Oracle (via LogMiner or XStream).
• Oracle LogMiner (Built-in Utility): A native Oracle Database feature that parses redo logs and archive logs without requiring separate licensing.
• Oracle GoldenGate (Commercial / Enterprise): Oracle's enterprise-grade replication suite. Unlike Debezium, GoldenGate is proprietary licensed commercial software.

Java Implementation with Debezium Embedded Engine (2.x+)

While running Debezium inside Kafka Connect is recommended for production scale, you can also embed the Debezium engine directly into a Java application:

import io.debezium.config.Configuration;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import org.apache.kafka.connect.source.SourceRecord;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class OracleCdcPipeline {

    public static void main(String[] args) {
        // Build modern Debezium 2.x configuration
        Configuration config = Configuration.create()
            .with("name", "oracle-cdc-engine")
            .with("connector.class", "io.debezium.connector.oracle.OracleConnector")
            .with("tasks.max", "1")
            .with("database.hostname", "oracle-server.internal")
            .with("database.port", "1521")
            .with("database.user", "c##dbzuser")
            .with("database.password", "dbzpassword")
            .with("database.dbname", "ORCLCDB")
            .with("database.pdb.name", "ORCLPDB1")
            .with("topic.prefix", "inventory")
            .with("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory")
            .with("schema.history.internal.file.filename", "/var/log/debezium/schema_history.dat")
            .build();

        // Create the Debezium Engine
        DebeziumEngine<RecordChangeEvent<SourceRecord>> engine = DebeziumEngine
            .create(ChangeEventFormat.of(Connect.class))
            .using(config.asProperties())
            .notifying(OracleCdcPipeline::handleRecord)
            .build();

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(engine);

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                engine.close();
                executor.shutdown();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }));
    }

    private static void handleRecord(RecordChangeEvent<SourceRecord> event) {
        SourceRecord record = event.record();
        System.out.printf("CDC Event Captured on Topic [%s]: %s%n", record.topic(), record.value());
    }
}