Database Integration & CDC #

Connecting operational relational databases (RDBMS) with Apache Kafka is one of the most common data integration scenarios in the industry. Transaction data entering the main database must be immediately routed to downstream systems for real-time processing. There are two main ways to do this integration using Kafka Connect: the traditional polling-query approach using the JDBC Source Connector, and the modern transaction-based approach using Change Data Capture (CDC). Although the polling query approach looks easy at first, it has fundamental limitations critical for large-scale operations. This article will dissect the weaknesses of the polling query approach, explain the basic concepts of transaction log-based CDC, describe the Debezium engine architecture, and unpack schema evolution handling when database table structures change.

Limitations of the Traditional Polling Query Approach (JDBC Source Connector) #

The JDBC Source Connector works by periodically sending SELECT SQL queries (for example every 5 seconds) to the target database. To detect new or updated rows, the connector relies on marker columns in the table, like continuously increasing numeric columns (incrementing id) or update timestamp marker columns (timestamp).

Although simple, this polling approach has four critical design weaknesses:

1. Cannot Detect Delete Operations #

This is the JDBC Source’s biggest weakness. When a data row is deleted from the database via a SQL DELETE command, that row disappears from the physical table. Because the row no longer exists, the JDBC Connector’s periodic SELECT queries will never detect it. As a result, downstream systems reading data from Kafka never know that row was deleted, causing data drift.

2. Heavy Query Overhead on the Main Database #

Continuously running SELECT queries against large tables with suboptimal indexes burdens the operational database’s CPU and I/O. This can slow down main customer business transactions (application database contention).

3. Loss of Intermediate Changes (Intermediate Updates) #

If a data row is updated several times between two polling intervals, the JDBC Connector only captures the row’s final state when the query runs. The entire history of intermediate changes between polling intervals is lost.

4. Dependence on Source Schema Modifications #

This approach requires a consistent last_updated_timestamp column in every table. Often, developer teams are forced to modify legacy schemas and create additional indexes just to support these polling needs.

What Is Log-Based Change Data Capture (CDC)? #

Log-based Change Data Capture (CDC) is a modern data integration technique reading every data change directly from the database’s binary transaction log.

Every modern relational database has an internal transaction log (like the Write-Ahead Log / WAL in PostgreSQL, the Binlog in MySQL, or the Redo Log in Oracle). This log is written sequentially (append-only) before data is written to physical tables to guarantee transaction durability (ACID).

sequenceDiagram
    participant App as Business Application
    participant DB as Database Engine
    participant Log as Transaction Log (WAL/Binlog)
    participant CDC as Debezium CDC Task
    participant Kafka as Kafka Broker

    App->>DB: Run SQL (INSERT/UPDATE/DELETE)
    Note over DB: Validate the transaction
    DB->>Log: Write the change to the Log (WAL/Binlog)
    DB-->>App: Confirm Transaction Success
    Note over DB: Write data to physical tables asynchronously
    
    CDC->>Log: Read the transaction log byte stream (Non-blocking)
    Log-->>CDC: Send binary change events
    Note over CDC: Convert binary to SourceRecord
    CDC->>Kafka: Publish events to the Kafka Topic

Why Is Log-Based CDC Superior? #

• Non-Blocking and Very Lightweight: CDC doesn’t send SELECT queries to database tables. It only asynchronously reads the transaction log files already on disk, so the overhead on operational database performance is almost zero.
• Captures All Events (Including DELETE): Every DELETE operation is clearly written in the transaction log. CDC can capture this delete event and send it to Kafka, usually as a special message with a null value (Tombstone) to tell downstream systems to delete the related data.
• Captures Intermediate Changes: Because every transaction write is recorded in the log sequentially, CDC guarantees no data change, no matter how small, is missed.
• No Schema Changes Needed: The database doesn’t need additional timestamp columns or indexes on source tables.

Initial Snapshotting Mechanism Details #

When Debezium first runs on a production database that has been operating for years, there are millions of historical data rows already stored in the database tables. Active binary transaction logs (like the WAL in PostgreSQL or the Binlog in MySQL) are usually configured to be automatically cleaned after a certain period (for example 7 days) to save disk space.

This means the active transaction log no longer stores the entire data history since the database was first created. Therefore, Debezium needs a special mechanism to align the initial data state through Initial Snapshotting.

Incremental Snapshot Workflow #

By default, Debezium runs the initial snapshot mode with a very orderly step sequence:

1. Last Offset Check: Debezium checks whether offset coordinates for this database are already registered in the connect-offsets topic. If so, the snapshot process is skipped, and Debezium directly switches to reading the transaction log from that offset.
2. Locking (Mode-Dependent): Debezium does schema read locking on the source database. In MySQL, Debezium by default obtains a global read lock (FLUSH TABLES WITH READ LOCK) to take a consistent latest Binlog coordinate. In PostgreSQL, Debezium uses the Exported Snapshot mechanism built into the SERIALIZABLE isolation-level transaction to avoid global locking that disrupts database writes.
3. DDL Schema Reading: Debezium reads the entire table structure schema in the integration scope and registers it to the Schema Registry.
4. Data Dumping (Mass Read): Debezium runs the SELECT * FROM table query efficiently (using streaming result sets) to copy the final data state from all tables at that moment. This data is sent to the Kafka topic with the op: r (read) operation code.
5. Lock Release and Log Transition: After all tables are copied, the read lock is released. Debezium records the binary transaction log coordinate where it started the snapshot, then smoothly transitions to real-time binary stream reading mode to process transactions occurring after the snapshot started.

The Debezium Architecture as the Main CDC Engine #

Debezium is a leading open-source CDC connector collection built on the Kafka Connect framework. Debezium provides special connector plugins for various popular databases like PostgreSQL, MySQL, SQL Server, Oracle, MongoDB, and Cassandra.

How Debezium Works #

Debezium acts like a replica instance of the source database. For example, the Debezium MySQL Connector registers itself to the MySQL server as a replication slave server using the MySQL binary replication protocol. The MySQL server then streams log change events directly to Debezium in real-time.

Anatomy of a Debezium Event Payload #

Every data change event sent by Debezium to Kafka has a detailed JSON/Avro structure including the state before and after the data changed, plus source metadata information.

Here’s a real example of a JSON payload from the Debezium PostgreSQL Connector when an UPDATE operation happens on the customers table:

{
  "schema": { ... },
  "payload": {
    "before": {
      "id": 102,
      "name": "Budi Santoso",
      "email": "[email protected]",
      "updated_at": 1672531190000
    },
    "after": {
      "id": 102,
      "name": "Budi Santoso",
      "email": "[email protected]",
      "updated_at": 1672531205000
    },
    "source": {
      "version": "2.1.2.Final",
      "connector": "postgresql",
      "name": "postgres-prod",
      "ts_ms": 1672531205120,
      "db": "inventory",
      "schema": "public",
      "table": "customers",
      "txId": 524,
      "lsn": 24598230
    },
    "op": "u",
    "ts_ms": 1672531205250
  }
}

Payload Field Explanation: #

• before: The data row state before the transaction ran. Useful for consumers wanting to compare old and new values.
• after: The data row state after the transaction successfully ran.
• source: Very rich source database metadata, containing the database name, table name, internal transaction ID (txId), transaction timestamp in the database (ts_ms), and even the log sequence number (lsn).
• op: The operation code triggering this event. Its values are:
  • c (create) for INSERT operations
  • u (update) for UPDATE operations
  • d (delete) for DELETE operations
  • r (read) for initial snapshot reads.

Debezium CDC Pipeline Performance Tuning #

To ensure the change rate in the main database streams to Kafka without high latency, we need to adjust several important performance tuning parameters in the Debezium task configuration:

# Maximum number of log change events read from the internal queue
# in one delivery cycle to Kafka Connect. Default: 2048.
max.batch.size=4096

# Maximum capacity of the internal queue holding messages before writing to Kafka.
# This value must always be larger than max.batch.size (e.g., 2x or 3x).
max.queue.size=12288

# Maximum pause duration (in milliseconds) for the Debezium poller thread
# to wait for new transaction log events to enter the queue before sending.
poll.interval.ms=500

# Determines how high-precision numeric data types (Decimal) are processed.
# Set to 'double' or 'string' to avoid complex Java BigDecimal binary overhead.
decimal.handling.mode=double

Schema Evolution Handling #

In long-term operations, database table structures will definitely change (DDL changes), like adding new columns, removing old columns, or modifying column data types. The process of handling these changes in distributed data pipelines is called Schema Evolution.

If we don’t manage schema evolution correctly, a simple PostgreSQL database modification can immediately trigger cascade failures on all downstream consumers surprised by the new data format.

1. Debezium Integration with Schema Registry #

To manage schema evolution safely, we must use the Confluent Schema Registry together with Avro or Protobuf serialization data formats.

When the database schema changes:

1. Debezium detects the DDL structure change from the transaction log.
2. Debezium assembles the new internal data schema.
3. Debezium sends that new schema to the Schema Registry for validation.
4. The Schema Registry checks the configured schema compatibility rules. If valid, the registry assigns a new schema ID.
5. Debezium publishes the new binary message to Kafka with the new schema ID.

2. Choosing Schema Compatibility Rules #

The Schema Registry provides several compatibility levels determining how schemas may change:

• BACKWARD Compatibility: Consumers using the new schema are guaranteed to still read data written with the old schema. This is the default and safest mode if we want to upgrade consumers first before producers (Debezium).
  • Rules: New columns may only be added if they’re optional (having default values or nullable). Old columns may not be removed unless they have default values.
• FORWARD Compatibility: Consumers using the old schema are guaranteed to still read data written with the new schema.
  • Rules: Old columns may be removed. New columns may not be added unless they’re optional.
• FULL Compatibility: Guarantees two-way compatibility (a combination of Backward and Forward).

Recommended Debezium Production Configuration #

Here are the important parameters we must set to maintain schema stability in Debezium:

# Ignoring database DDL columns unsupported by Kafka Connect
# so the task doesn't instantly crash when database admins run custom DDL queries
database.history.skip.unparseable.ddl=true

# Storing database DDL change history to a special Kafka topic
# for historical schema recovery if the task is restarted from scratch
database.history.kafka.topic=schema-changes.inventory
database.history.kafka.bootstrap.servers=kafka-1:9092

JDBC Source vs CDC (Debezium) Comparison #

To help us choose the right database integration solution, here’s a comprehensive feature comparison matrix:

Summary #

• Polling Weakness — The JDBC Source Connector uses periodic SELECT queries burdening the operational database and unable to detect data deletion (DELETE) operations.
• Log-Based CDC — CDC reads the database binary transaction log (like WAL or Binlog) non-blockingly, offering minimal overhead and recording every transaction detail.
• Debezium Payload — Debezium payloads are very comprehensive, recording data before (before), data after (after), source metadata (source), and the operation type (op).
• Initial Snapshotting — The snapshot mechanism safely copies the initial data state from the database before Debezium switches to processing active transaction log bytes.
• Schema Evolution — Schema evolution must be managed using a Schema Registry to guarantee data format compatibility when database table structures change.

← Previous: Standalone vs Distributed Mode Next: File & Object Storage Integration →