Streaming MySQL Changes to Kafka with Debezium

Change data capture (CDC) turns a database’s write-ahead log into an event stream, and it is the foundation of most event-driven and real-time data architectures. Streaming MySQL changes reliably into Apache Kafka requires understanding the Kafka Connect runtime, schema management, and a handful of operational gotchas that only surface in production. This guide walks through capturing MySQL row-level changes with the Debezium MySQL connector, with the configuration written for Debezium 2.x, the current stable line.

The Schema Registry & Connect Ecosystem is the backbone of this integration. Debezium is a Kafka Connect source connector, so how Connect distributes tasks, stores offsets, and handles failures determines how the CDC pipeline behaves. This guide covers the full lifecycle—initial snapshot, ongoing binlog streaming, schema evolution, and recovery—and calls out the property renames and error-handling limits that trip up operators migrating from older Debezium releases.

Debezium MySQL Connector Architecture

The connector reads MySQL’s binary log (binlog) as its source of truth. On first start it takes an initial consistent snapshot of the configured tables. With the default locking mode, it briefly holds a global read lock to read schema and capture binlog coordinates, then releases the lock and reads table rows inside a REPEATABLE READ transaction so that concurrent writes do not corrupt the snapshot. When the snapshot finishes, the connector transitions into streaming mode and tails binlog events.

The MySQL connector uses a single task to read the binlog (tasks.max higher than 1 has no effect on streaming), because the binlog is a single ordered stream and parallel readers would break ordering. The task uses a dedicated MySQL connection and persists its position—binlog file/offset, or GTID set when GTIDs are enabled—through Kafka Connect’s offset storage, which is itself a compacted Kafka topic managed by the worker.

# Debezium MySQL connector config (Debezium 2.x property names)
name=mysql-inventory-connector
connector.class=io.debezium.connector.mysql.MySqlConnector
database.hostname=mysql-primary.example.com
database.port=3306
database.user=debezium
database.password=${DEBEZIUM_DB_PASSWORD}
database.server.id=184054
topic.prefix=mysql_inventory
database.include.list=inventory
table.include.list=inventory.customers,inventory.orders
schema.history.internal.kafka.bootstrap.servers=kafka-broker-1:9092,kafka-broker-2:9092
schema.history.internal.kafka.topic=schema-changes.mysql_inventory
include.schema.changes=true
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081

Property rename (Debezium 1.x → 2.x): database.server.name is now topic.prefix, and database.history.kafka.bootstrap.servers / database.history.kafka.topic are now schema.history.internal.kafka.bootstrap.servers / schema.history.internal.kafka.topic. The old names were removed in 2.0; configs that still use them will fail validation. Note the connector needs both the schema-history bootstrap servers and topic—omitting the bootstrap servers is a common 2.x migration mistake.

Each row change becomes one Kafka record: the key holds the primary-key columns, and the value holds the before and after row state plus a source metadata block. Data change events are written to topics named <topic.prefix>.<database>.<table> (for example mysql_inventory.inventory.customers). Schema change events, when include.schema.changes=true, go to a single topic named exactly <topic.prefix>. This naming informs how you size and place workers in Kafka Connect Architecture and Deployment Modes.

Configuring MySQL for Debezium

MySQL must be prepared for CDC before the connector starts. The hard requirement is a row-format binlog: statement-based and mixed-format binlogs do not carry the full row image Debezium needs. The connector’s MySQL account needs SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT.

-- Verify current binlog configuration
SHOW VARIABLES LIKE 'log_bin%';
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';

-- Required parameters in my.cnf
-- [mysqld]
-- server-id                   = 223344
-- log_bin                     = /var/lib/mysql/mysql-bin
-- binlog_format               = ROW
-- binlog_row_image            = FULL
-- binlog_expire_logs_seconds  = 864000   -- 10 days (MySQL 8.0.1+)

-- Create the Debezium user with the minimum required privileges
CREATE USER 'debezium'@'%' IDENTIFIED BY 'strong_password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT
  ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

binlog_row_image=FULL ensures every column appears in the binlog for updates and deletes, not just the changed columns—without it, before images and unchanged columns are missing. Binlog retention governs how long the connector can be offline: if it is stopped longer than the binlog is kept, its saved position is purged and it must re-snapshot. On MySQL 8.0.1+ use binlog_expire_logs_seconds (the older expire_logs_days is deprecated); on 5.7 use expire_logs_days.

LOCK TABLES for hosted MySQL: On Amazon RDS/Aurora and other managed instances that disallow a global read lock (FLUSH TABLES WITH READ LOCK), Debezium falls back to table-level locks for the snapshot. In that case the account also needs LOCK TABLES. If you cannot grant it, set snapshot.locking.mode=none and ensure no DDL runs during the snapshot.

For HA MySQL using GTID replication, Debezium can track GTID sets so that after a failover it resumes from the correct transaction without manual binlog-coordinate surgery. Enable gtid_mode=ON and enforce_gtid_consistency=ON in MySQL; on the connector, scope which GTID sources to follow with gtid.source.includes / gtid.source.excludes (regular expressions matched against source UUIDs).

Schema Management and the Schema Registry

With the Avro converter, Debezium derives an Avro schema for each table’s key and value and registers them in the Schema Registry, giving you a typed, evolvable contract between producers and consumers. The value schema nests the before, after, and source structures.

Schema evolution is the operationally hard part. When DDL alters a captured table, Debezium parses the statement from the binlog, records it in the schema history topic, and emits subsequent events with the new schema. The Schema Registry then enforces the subject’s configured compatibility rule. For CDC value subjects, BACKWARD compatibility is a common default: it lets consumers on the previous schema read records written with the new one, which covers the typical case of adding nullable fields.

# Check a candidate value schema against the latest registered version
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema":"{\"type\":\"record\",\"name\":\"customers\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"email\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"phone\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}' \
  http://schema-registry:8081/compatibility/subjects/mysql_inventory.inventory.customers-value/versions/latest

# Inspect the current compatibility level for the subject
curl http://schema-registry:8081/config/mysql_inventory.inventory.customers-value

In practice the connector registers schemas automatically; you rarely POST them by hand, so test compatibility first rather than registering blindly. Watch for failed compatibility checks—they surface as serialization errors that stop the task. The classic trap is a column rename: Debezium sees a drop plus an add, which breaks consumers that referenced the old field name. The Debezium MySQL schema change topic documentation details how these events are structured.

Deploying and Operating the Connector in Production

Debezium runs as a plugin inside a Kafka Connect worker, and the worker’s settings shape connector behavior. Two worker-level properties matter most here: offset.flush.interval.ms, which controls how often Connect commits source offsets (the binlog position), and task.shutdown.graceful.timeout.ms, which bounds how long the worker waits for a task to stop during a rebalance. These are set in the worker config, not the connector.

{
  "name": "mysql-inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql-primary.example.com",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "${DEBEZIUM_DB_PASSWORD}",
    "database.server.id": "184054",
    "topic.prefix": "mysql_inventory",
    "database.include.list": "inventory",
    "table.include.list": "inventory.customers,inventory.orders",
    "schema.history.internal.kafka.bootstrap.servers": "kafka-broker-1:9092,kafka-broker-2:9092",
    "schema.history.internal.kafka.topic": "schema-changes.mysql_inventory",
    "include.schema.changes": "true",
    "snapshot.mode": "initial",
    "snapshot.locking.mode": "minimal",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Do not put worker settings in the connector config. offset.storage.topic, config.storage.topic, and status.storage.topic are distributed-worker properties (set in the worker config, not per connector). Likewise, the legacy internal.key.converter / internal.value.converter properties were deprecated in Kafka 2.0 (KIP-174) and removed in Kafka 3.0 (KIP-738); Connect now uses JsonConverter for internal topics unconditionally. Putting any of these in a connector config is invalid and will be rejected or ignored.

snapshot.locking.mode controls lock duration. The default minimal holds the global read lock only while reading schema and binlog coordinates, then relies on the REPEATABLE READ transaction for the rest of the snapshot. On write-heavy instances even this brief lock can stall writes; extended blocks writes for the whole snapshot (safest, most disruptive), and none takes no locks at all (only safe if no DDL runs during the snapshot). For snapshot.mode, use initial for a one-time full snapshot then stream, no_data to capture only the schema and stream from the current position (this replaces the deprecated schema_only), and never to stream from the current binlog position without recording schema up front. Use recovery to rebuild a lost or corrupted schema history topic.

Monitor the connector through JMX. Debezium exposes streaming metrics under the MBean debezium.mysql:type=connector-metrics,context=streaming,server=<topic.prefix>, including MilliSecondsBehindSource (lag between a change being committed and Debezium reading it) and TotalNumberOfEventsSeen (throughput). Snapshot progress lives under context=snapshot and schema-history health under context=schema-history. The Debezium monitoring documentation lists the full MBean set, and Connect’s own framework metrics cover task state and rebalance activity.

Handling Failures and Recovering State

Failures split into transient (recover by retry) and persistent (need an operator). Transient cases—brief network blips, a MySQL restart, a Kafka leader election—are handled by the Connect framework’s retry logic plus Debezium’s own database-reconnect handling.

Use the real Connect error properties. The framework retry knobs introduced by KIP-298 (Kafka 2.0) are errors.retry.timeout (total time budget for retries; negative means retry forever) and errors.retry.delay.max.ms (cap on the exponential backoff). There is no errors.retry.max.attempts property—it does not exist in Kafka Connect.

A critical limitation: the dead letter queue (errors.deadletterqueue.*) is a sink-connector-only feature. Debezium MySQL is a source connector, so a DLQ does not apply to it—pointing those properties at it has no effect. The source-side equivalents are Debezium’s own handlers:

# Source-connector error handling (Debezium MySQL)
# Connect framework retry budget (KIP-298)
errors.retry.timeout=300000          # retry transient failures for up to 5 minutes
errors.retry.delay.max.ms=60000      # cap backoff between retries at 60s
errors.log.enable=true
errors.log.include.messages=true

# Debezium-specific handling of unparseable / inconsistent events
event.processing.failure.handling.mode=fail   # fail | warn | skip
inconsistent.schema.handling.mode=fail         # fail | warn | skip

event.processing.failure.handling.mode governs what happens when the connector hits a binlog event it cannot process: fail stops the connector (the safe default), warn logs and skips, skip silently skips. Set it to warn/skip only when dropping events is acceptable, because it risks silent data loss. inconsistent.schema.handling.mode covers events for a table whose schema the connector does not know.

When a connector cannot recover, decide between restarting from the saved offset (preferred—no re-read) and re-snapshotting. If the binlog position is purged or the schema history topic is lost, you need a new snapshot.

# Restart a failed connector (and its tasks)
curl -X POST "http://connect-worker:8083/connectors/mysql-inventory-connector/restart?includeTasks=true&onlyFailed=true"

# Read the full current config, then re-PUT it (PUT replaces the ENTIRE config)
curl http://connect-worker:8083/connectors/mysql-inventory-connector/config
# ...edit snapshot.mode in that JSON, then:
curl -X PUT -H "Content-Type: application/json" \
  --data @connector-config.json \
  http://connect-worker:8083/connectors/mysql-inventory-connector/config

PUT /connectors/{name}/config replaces the entire configuration. You cannot send just {"snapshot.mode":"initial"}—that wipes every other setting and the connector fails. GET the current config, change the one field, and PUT the whole document back. For re-snapshotting specific tables without a restart, prefer an ad hoc incremental snapshot via Debezium’s signaling mechanism, which runs in chunks while streaming continues uninterrupted.

The schema history topic (schema.history.internal.kafka.topic) is essential recovery state: it stores every captured DDL statement so Debezium can reconstruct table schemas at any binlog position. Give it a replication factor of at least 3 and treat its loss as forcing a recovery-mode run or a full re-snapshot. The Debezium MySQL connector documentation covers recovery in depth.

Performance Tuning and Scaling Considerations

The binlog reader is single-task, but several properties move throughput and latency. max.batch.size and max.queue.size set how many events are batched per producer flush and buffered in memory; raising them lifts throughput at the cost of memory and end-to-end latency.

For large initial snapshots, snapshot.fetch.size sets the JDBC fetch size per round-trip; bigger reduces query count but pressures memory on both MySQL and the worker. min.row.count.to.stream.results is the row-count threshold above which Debezium switches a snapshot query to a cursor-based streaming result set instead of buffering the whole result.

# Performance tuning parameters
max.batch.size=4096
max.queue.size=16384
snapshot.fetch.size=10240
min.row.count.to.stream.results=1000
poll.interval.ms=500
connect.timeout.ms=30000

Network latency between the worker and MySQL dominates streaming lag—place workers in the same availability zone as the MySQL primary. For multi-datacenter setups you can snapshot from a local read replica, but binlog streaming should target the primary (or a replica whose binlog and GTID set you control). Debezium keys data-change records by the source primary key, ordering all changes to a given row; for skewed keys you can reroute or repartition with Single Message Transforms (for example ByLogicalTableRouter), at some processing cost.

Operational Readiness Checklist

Before promoting a Debezium MySQL pipeline to production, validate:

  • Binlog retention — binlog_expire_logs_seconds (MySQL 8.0.1+) or expire_logs_days (5.7) exceeds the maximum planned connector downtime with margin (e.g. +48h).
  • Snapshot privileges — on RDS/Aurora or any instance without global read locks, grant LOCK TABLES or set snapshot.locking.mode=none (with no concurrent DDL).
  • Schema compatibility — set the Schema Registry compatibility for value subjects (commonly BACKWARD) and rehearse a column rename, since Debezium treats it as drop+add.
  • Monitoring — alert on MilliSecondsBehindSource exceeding your lag SLO and on connector/task state != RUNNING.
  • Error handling — set event.processing.failure.handling.mode and inconsistent.schema.handling.mode deliberately; remember a DLQ does not apply to this source connector.
  • Schema history topic — replication factor >= 3; it is automatically compacted, and its loss forces recovery mode or a re-snapshot. Document that procedure.
  • Security — TLS for all hops (Connect↔MySQL, Connect↔Kafka, Connect↔Schema Registry). Keep credentials in a secrets manager and inject them with config providers / env substitution (e.g. ${DEBEZIUM_DB_PASSWORD}), never plaintext in the connector config.
  • Testing — exercise representative DDL (add column, rename, type change) end to end in staging before production.
Back to Kafka Connect Architecture And Deployment Modes