Debezium MySQL Source Connector - AutoMQ Documentation

Overview

Debezium MySQL Source Connector reads MySQL table schemas, existing rows, and row-level and DDL changes from the binlog, then publishes them to Kafka as Kafka Connect SourceRecords. It usually takes a snapshot on the first run and then continues streaming from the snapshot boundary. By default, changes from each table go to a Topic organized by Topic prefix, database name, and table name. It is suitable for continuously sending MySQL change events to Kafka for downstream real-time processing, cache synchronization, auditing, or data lake processing.

Prerequisites

MySQL must have binary logging enabled and use binlog_format=ROW and binlog_row_image=FULL; Binlog retention should cover expected recovery time and synchronization delay.
The connection user needs SELECT privileges on snapshot tables and the REPLICATION SLAVE and REPLICATION CLIENT privileges required for replication and position reads. Database discovery and snapshot locking may also require SHOW DATABASES and RELOAD; table-lock fallback also requires LOCK TABLES. Managed MySQL services may use different privilege names.
database.server.id must be unique among currently running MySQL clients. When using GTID or read-only incremental snapshots, the source server must also retain the required GTID history.
Kafka Schema History uses a dedicated single-partition Topic that retains the complete DDL history and permits the Connector to read, write, and create the Topic using the corresponding Kafka client configuration.

License

Licensed under Apache License 2.0.

Quick Start

Prepare the Connect Cluster, Kafka, and the MySQL database to capture. Confirm that the Connector can access the database, Kafka, and the Schema History Topic, and complete the required authorization. For Connector creation and management, see Manage Connectors. The following configuration takes an initial snapshot of existing tables and then continuously reads the binlog; replace the environment values in angle brackets.

name=debezium-mysql-source
connector.class=io.debezium.connector.mysql.MySqlConnector
topic.prefix=mysql-prod
database.hostname=<database-host>
database.port=3306
database.user=<database-user>
database.password=<database-password>
database.server.id=5401
schema.history.internal.kafka.topic=<schema-history-topic>
schema.history.internal.kafka.bootstrap.servers=<kafka-bootstrap-servers>

topic.prefix, the Schema History Topic, and database.server.id should each remain unique across Connectors; provide the password through a secure configuration method supported by the deployment environment.

Configuration

Identity

`topic.prefix`

Creates a Topic namespace for this MySQL server or cluster and must be unique across Connectors.

Type: STRING
Default: None
Importance: High
Valid Values / Notes: only alphanumeric, hyphen, dot, and underscore; Required

Database Connection

`database.hostname`

The resolvable hostname or IP address of the MySQL server.

Type: STRING
Default: None
Importance: High
Valid Values / Notes: ^[a-zA-Z0-9-_.]+$; Required

`database.port`

The MySQL server port.

Type: INT
Default: 3306
Importance: High
Valid Values / Notes: integer

`database.user`

The username used to connect to MySQL.

Type: STRING
Default: None
Importance: High
Valid Values / Notes: Required

`database.password`

The password used to connect to MySQL.

Type: PASSWORD
Default: None
Importance: High
Valid Values / Notes: Provide a value of the specified type; no additional constraints are listed.

`database.query.timeout.ms`

Database query timeout in milliseconds.

Type: INT
Default: 600000
Importance: Low
Valid Values / Notes: integer; 0 means no limit

`database.initial.statements`

SQL executed when a JDBC connection is established; intended for session settings, not data modification.

Type: STRING
Default: None
Importance: Low
Valid Values / Notes: semicolon-separated SQL; use ;; for a literal semicolon

`database.server.id`

The unique numeric ID used when the Connector joins the MySQL cluster as a binlog client.

Type: LONG
Default: None
Importance: High
Valid Values / Notes: positive long; unique among running database clients; Required

`database.server.id.offset`

Server ID offset for parallel snapshot connections.

Type: LONG
Default: 10000
Importance: High
Valid Values / Notes: long; Depends on: snapshot.max.threads

SSL/TLS

`database.ssl.keystore`

Path to the client keystore file.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: Depends on: database.ssl.keystore.password

`database.ssl.keystore.password`

Password for the client keystore.

Type: PASSWORD
Default: None
Importance: Medium
Valid Values / Notes: Depends on: database.ssl.keystore

`database.ssl.truststore`

Path to the truststore used to verify the server certificate.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: Depends on: database.ssl.truststore.password

`database.ssl.truststore.password`

Password used to verify truststore integrity.

Type: PASSWORD
Default: None
Importance: Medium
Valid Values / Notes: Depends on: database.ssl.truststore

Database Connection

`database.jdbc.driver`

JDBC driver class for MySQL connections.

Type: CLASS
Default: com.mysql.cj.jdbc.Driver
Importance: Low
Valid Values / Notes: valid class name

`database.protocol`

Connection protocol used by the JDBC driver.

Type: STRING
Default: jdbc:mysql
Importance: Low
Valid Values / Notes: Depends on: database.jdbc.driver

SSL/TLS

`database.ssl.mode`

Controls whether the database connection is encrypted and whether certificates and host identity are verified.

Type: STRING
Default: preferred
Importance: Medium
Valid Values / Notes: disabled, preferred, required, verify_ca, verify_identity; Depends on: database.ssl.keystore, database.ssl.truststore

Error Handling

`event.processing.failure.handling.mode`

Controls whether processing stops, logs and skips, or directly skips corrupted or unparseable events.

Type: STRING
Default: fail
Importance: Medium
Valid Values / Notes: fail, warn, ignore, skip

Throughput and Buffering

`max.batch.size`

The maximum number of SourceRecords returned to Kafka Connect in each batch.

Type: INT
Default: 2048
Importance: Medium
Valid Values / Notes: positive integer; Depends on: max.queue.size

`max.queue.size`

The maximum number of change events that have been read but not yet forwarded; must be greater than max.batch.size.

Type: INT
Default: 8192
Importance: Medium
Valid Values / Notes: positive; strictly greater than max.batch.size; Depends on: max.batch.size

`poll.interval.ms`

The time to wait before the next poll when no new events are available.

Type: LONG
Default: 500
Importance: Medium
Valid Values / Notes: positive integer

`max.queue.size.in.bytes`

Byte limit for the change-event queue, applied alongside the record-count limit.

Type: LONG
Default: 0
Importance: Medium
Valid Values / Notes: non-negative; 0 disables byte bound

Events

`provide.transaction.metadata`

Whether to generate transaction metadata and event counts.

Type: BOOLEAN
Default: false
Importance: Low
Valid Values / Notes: boolean; Depends on: transaction.metadata.factory

`skipped.operations`

Specifies the operation types to skip during streaming synchronization.

Type: LIST
Default: t
Importance: Low
Valid Values / Notes: r,c,u,d,t,none; none cannot be combined with another operation

Snapshots

`snapshot.delay.ms`

Delay before starting a snapshot, in milliseconds.

Type: LONG
Default: 0
Importance: Low
Valid Values / Notes: non-negative long

`streaming.delay.ms`

Delay between snapshot completion and streaming, in milliseconds.

Type: LONG
Default: 0
Importance: Low
Valid Values / Notes: non-negative long

`snapshot.include.collection.list`

Table-name regular expressions limiting the snapshot scope.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of regular expressions; used to select snapshot tables; Depends on: snapshot.mode

`snapshot.fetch.size`

Number of rows fetched per batch by snapshot queries.

Type: INT
Default: None
Importance: Medium
Valid Values / Notes: non-negative integer

`snapshot.max.threads`

The maximum number of threads used by the snapshot; increasing this value does not increase the number of Kafka Connect Tasks.

Type: INT
Default: 1
Importance: Medium
Valid Values / Notes: positive integer; Depends on: database.server.id.offset

`snapshot.mode.custom.name`

Name of the custom snapshot mode implementation.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: non-empty when snapshot.mode=custom; Depends on: snapshot.mode

`snapshot.mode.configuration.based.snapshot.data`

Whether configuration_based mode snapshots row data.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; Depends on: snapshot.mode=configuration_based

`snapshot.mode.configuration.based.snapshot.schema`

Whether configuration_based mode snapshots table schemas.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; Depends on: snapshot.mode=configuration_based

`snapshot.mode.configuration.based.start.stream`

Whether configuration_based mode starts streaming.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; Depends on: snapshot.mode=configuration_based

`snapshot.mode.configuration.based.snapshot.on.schema.error`

Whether configuration_based mode requests a snapshot on schema-history errors.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; Depends on: snapshot.mode=configuration_based

`snapshot.mode.configuration.based.snapshot.on.data.error`

Whether configuration_based mode requests a snapshot when the log position is unavailable.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; Depends on: snapshot.mode=configuration_based

Error Handling

`retriable.restart.connector.wait.ms`

Wait before restarting after a retriable error, in milliseconds; not all database errors are retriable.

Type: LONG
Default: 10000
Importance: Low
Valid Values / Notes: positive long

Throughput and Buffering

`query.fetch.size`

JDBC query fetch size during streaming.

Type: INT
Default: 0
Importance: Medium
Valid Values / Notes: non-negative; 0 uses JDBC default

Error Handling

`errors.max.retries`

Maximum retry count for retriable errors.

Type: INT
Default: -1
Importance: Low
Valid Values / Notes: -1 unlimited, 0 disabled, positive number of retries

Incremental Snapshots

`incremental.snapshot.watermarking.strategy`

Watermark strategy used in the signal table to open and close incremental snapshot windows.

Type: STRING
Default: INSERT_INSERT
Importance: Low
Valid Values / Notes: INSERT_INSERT, INSERT_DELETE; Depends on: signal.data.collection

Database Connection

`connection.validation.timeout.ms`

Database connection validation timeout in milliseconds.

Type: LONG
Default: 60000
Importance: Low
Valid Values / Notes: positive long

Runtime

`executor.shutdown.timeout.ms`

Executor shutdown timeout in milliseconds.

Type: LONG
Default: 4000
Importance: Medium
Valid Values / Notes: positive long

Integration

`openlineage.integration.enabled`

Whether to enable OpenLineage integration.

Type: BOOLEAN
Default: false
Importance: Low
Valid Values / Notes: boolean; Depends on: openlineage.integration.config.file.path

`openlineage.integration.config.file.path`

Path to the OpenLineage configuration file.

Type: STRING
Default: ./openlineage.yml
Importance: Low
Valid Values / Notes: Depends on: openlineage.integration.enabled

`openlineage.integration.job.namespace`

Namespace for the OpenLineage job.

Type: STRING
Default: None
Importance: Low
Valid Values / Notes: Depends on: openlineage.integration.enabled

`openlineage.integration.job.description`

Description of the OpenLineage job.

Type: STRING
Default: Debezium change data capture job
Importance: Low
Valid Values / Notes: Depends on: openlineage.integration.enabled

`openlineage.integration.job.tags`

Tags for the OpenLineage job.

Type: LIST
Default: None
Importance: Low
Valid Values / Notes: list of key=value pairs; Depends on: openlineage.integration.enabled

`openlineage.integration.job.owners`

Owner information for the OpenLineage job.

Type: LIST
Default: None
Importance: Low
Valid Values / Notes: list of key=value pairs; Depends on: openlineage.integration.enabled

Events

`extended.headers.enabled`

Whether to add extended event headers.

Type: BOOLEAN
Default: true
Importance: Low
Valid Values / Notes: boolean

Type Conversion

`decimal.handling.mode`

Controls the representation of DECIMAL/NUMERIC values in events.

Type: STRING
Default: precise
Importance: Medium
Valid Values / Notes: precise, string, double

Snapshots

`snapshot.lock.timeout.ms`

Snapshot lock acquisition timeout in milliseconds.

Type: LONG
Default: 10000
Importance: Medium
Valid Values / Notes: long; lock acquisition timeout; Depends on: snapshot.locking.mode

Database Connection

`connect.timeout.ms`

Database connection timeout in milliseconds.

Type: INT
Default: 30000
Importance: Medium
Valid Values / Notes: positive integer

`connect.keep.alive`

Whether to enable the binlog connection keepalive thread.

Type: BOOLEAN
Default: true
Importance: Low
Valid Values / Notes: boolean; Depends on: connect.keep.alive.interval.ms

`connect.keep.alive.interval.ms`

Binlog connection keepalive check interval in milliseconds.

Type: LONG
Default: 60000
Importance: Low
Valid Values / Notes: positive integer; Depends on: connect.keep.alive

`use.nongraceful.disconnect`

Whether to disconnect the binlog client without a graceful shutdown.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean

Snapshots

`snapshot.mode`

Determines when to take a snapshot and whether to continue streaming after the snapshot.

Type: STRING
Default: initial
Importance: Low
Valid Values / Notes: always, when_needed, initial, initial_only, never, configuration_based, custom, no_data, recovery, schema_only, schema_only_recovery; recovery is safe only when an offset already exists and no Schema changes occurred since the Connector was stopped; do not use it after post-shutdown Schema changes, and do not relax this condition based on DDL compatibility. Required binlogs must still be available; rebuilding Schema history cannot restore missing logs. when_needed does not handle Schema History errors or restore lost historical events. Depends on: snapshot.mode.custom.name, snapshot.mode.configuration.based.snapshot.data, snapshot.mode.configuration.based.snapshot.schema, snapshot.mode.configuration.based.start.stream

`snapshot.query.mode`

Selects the snapshot query implementation.

Type: STRING
Default: select_all
Importance: Low
Valid Values / Notes: select_all, custom; Depends on: snapshot.query.mode.custom.name

`snapshot.query.mode.custom.name`

Name of the custom snapshot query implementation.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: non-empty when snapshot.query.mode=custom; Depends on: snapshot.query.mode

Type Conversion

`bigint.unsigned.handling.mode`

Controls representation of BIGINT UNSIGNED values that exceed the signed range。

Type: STRING
Default: long
Importance: Medium
Valid Values / Notes: long, precise

`time.precision.mode`

Controls precision representation of time, date, and timestamp fields。

Type: STRING
Default: adaptive_time_microseconds
Importance: Medium
Valid Values / Notes: adaptive_time_microseconds, connect; adaptive is not accepted

`enable.time.adjuster`

Whether to adjust temporal values with two-digit years.

Type: BOOLEAN
Default: true
Importance: Low
Valid Values / Notes: boolean

`schema.name.adjustment.mode`

Adjusts schema names to meet serialization naming rules.

Type: STRING
Default: none
Importance: Low
Valid Values / Notes: avro, avro_unicode, none

Snapshots

`min.row.count.to.stream.results`

Table row-count threshold for streaming snapshot query results.

Type: INT
Default: 1000
Importance: Low
Valid Values / Notes: non-negative; 0 streams all results

Incremental Snapshots

`incremental.snapshot.chunk.size`

The number of records read in each incremental snapshot chunk.

Type: INT
Default: 1024
Importance: Medium
Valid Values / Notes: non-negative integer; Depends on: signal.data.collection

`incremental.snapshot.allow.schema.changes`

Whether Schema changes are allowed during incremental snapshots.

Type: BOOLEAN
Default: false
Importance: Low
Valid Values / Notes: boolean; primary-key changes are unsupported; Depends on: incremental.snapshot.chunk.size

Snapshots

`snapshot.locking.mode`

Controls the locking strategy used while reading the snapshot.

Type: STRING
Default: minimal
Importance: Low
Valid Values / Notes: extended, minimal, minimal_percona, minimal_percona_no_table_locks, none, or custom. minimal holds the global read lock while reading schemas and metadata, then relies on REPEATABLE READ and a compatible transactional storage engine to read rows; consistency is not guaranteed for arbitrary engines or concurrent DDL. none acquires no snapshot locks; its usage boundary is snapshot.mode=schema_only or schema_only_recovery, with external controls preventing concurrent DDL throughout the snapshot. Passing enum validation does not establish that other combinations are safe. custom also requires snapshot.locking.mode.custom.name.

Internal Schema History

`schema.history.internal`

The Schema History implementation class; the Kafka storage implementation is used by default.

Type: CLASS
Default: io.debezium.storage.kafka.history.KafkaSchemaHistory
Importance: Low
Valid Values / Notes: SchemaHistory implementation class; Kafka and file storage are supported. Depends on: schema.history.internal.kafka.topic, schema.history.internal.kafka.bootstrap.servers.

`schema.history.internal.skip.unparseable.ddl`

Whether to skip DDL statements that cannot be parsed; skipping them can leave Schema metadata incomplete.

Type: BOOLEAN
Default: false
Importance: Low
Valid Values / Notes: true or false; skipping DDL can leave later events without the correct field definitions; Depends on: schema.history.internal

`schema.history.internal.store.only.captured.tables.ddl`

Whether to store only the DDL of captured tables in internal Schema History.

Type: BOOLEAN
Default: false
Importance: Low
Valid Values / Notes: true or false; before expanding table filters, confirm that the required historical Schema remains available; Depends on: schema.history.internal, table.include.list, table.exclude.list

`schema.history.internal.store.only.captured.databases.ddl`

Whether to store only the DDL of captured databases in internal Schema History.

Type: BOOLEAN
Default: false
Importance: Low
Valid Values / Notes: true or false; before expanding database filters, confirm that the required historical Schema remains available; Depends on: schema.history.internal, database.include.list, database.exclude.list

Extensions

`converters`

Aliases for custom type converters.

Type: STRING
Default: None
Importance: Low
Valid Values / Notes: comma-separated prefixes; each prefix.type is a CustomConverter class

`post.processors`

Aliases for event post-processors.

Type: STRING
Default: None
Importance: Low
Valid Values / Notes: comma-separated prefixes; each prefix.type is a post-processor class

Events

`tombstones.on.delete`

Whether to send a same-key null-valued tombstone after a delete event.

Type: BOOLEAN
Default: true
Importance: Medium
Valid Values / Notes: boolean

Heartbeat

`heartbeat.interval.ms`

The interval for sending heartbeat events; 0 means that heartbeats are not sent.

Type: INT
Default: 0
Importance: Medium
Valid Values / Notes: non-negative; 0 disables; Depends on: heartbeat.topics.prefix

`heartbeat.topics.prefix`

Name prefix for heartbeat topics.

Type: STRING
Default: __debezium-heartbeat
Importance: Low
Valid Values / Notes: Depends on: heartbeat.interval.ms

Signals

`signal.data.collection`

The MySQL table that receives incremental snapshot or other signals.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: fully qualified data collection name; Depends on: signal.enabled.channels

`signal.poll.interval.ms`

Interval for checking new signals, in milliseconds.

Type: LONG
Default: 5000
Importance: Medium
Valid Values / Notes: positive integer; Depends on: signal.enabled.channels

`signal.enabled.channels`

The list of enabled signal channels.

Type: LIST
Default: source
Importance: Medium
Valid Values / Notes: channel names; source enabled by default; Depends on: signal.data.collection

Topic Naming

`topic.naming.strategy`

The strategy class used to generate data, Schema change, and auxiliary Topic names.

Type: CLASS
Default: io.debezium.schema.SchemaTopicNamingStrategy
Importance: Medium
Valid Values / Notes: TopicNamingStrategy implementation

Notifications

`notification.enabled.channels`

Enabled notification channels.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: notification channel names; Depends on: notification.sink.topic.name

`notification.sink.topic.name`

Kafka topic for notifications from the sink notification channel.

Type: STRING
Default: None
Importance: High
Valid Values / Notes: required when notification.enabled.channels contains sink; Depends on: notification.enabled.channels

Events

`transaction.metadata.factory`

Factory implementation class for transaction metadata.

Type: CLASS
Default: io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory
Importance: Low
Valid Values / Notes: TransactionMetadataFactory implementation; Depends on: provide.transaction.metadata

Monitoring

`custom.metric.tags`

Custom tags added to metrics.

Type: LIST
Default: None
Importance: Low
Valid Values / Notes: list of key=value pairs

Filtering

`column.include.list`

Keeps values for matching columns only; the value is a comma-separated list of regular expressions.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of regular expressions; mutually exclusive with column.exclude.list; Depends on: column.exclude.list

`column.exclude.list`

Excludes values for matching columns; cannot be configured together with column.include.list.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of regular expressions; mutually exclusive with column.include.list; Depends on: column.include.list

`table.include.list`

Captures only matching database tables; the value is a comma-separated list of regular expressions.

Type: LIST
Default: None
Importance: High
Valid Values / Notes: list of regular expressions; mutually exclusive with table.exclude.list; Depends on: table.exclude.list, database.include.list

`table.exclude.list`

Excludes matching database tables; cannot be configured together with table.include.list.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of regular expressions; mutually exclusive with table.include.list; Depends on: table.include.list

Key Mapping

`message.key.columns`

Defines the columns used for message keys for specified tables.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: semicolon-separated table:key expressions; must match connector key pattern

Snapshots

`snapshot.select.statement.overrides`

Tables whose snapshots use custom SELECT queries.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: comma-separated fully qualified tables; each table needs snapshot.select.statement.overrides.<DB>.<TABLE>; Depends on: snapshot.mode

Field Masking

`column.mask.hash.([^.]+).with.salt.(.+)`

Hashes matching columns using the specified algorithm and salt.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: dynamic key; regex column selector plus hash algorithm and salt

`column.mask.with.(d+).chars`

Replaces matching column values with a mask of the specified length.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: dynamic key; integer mask length and column regex list

`column.truncate.to.(d+).chars`

Truncates matching column values to the specified character count.

Type: INT
Default: None
Importance: Medium
Valid Values / Notes: dynamic key; integer truncation length

Schema Events

`include.schema.changes`

Whether to publish Schema change events.

Type: BOOLEAN
Default: true
Importance: Medium
Valid Values / Notes: boolean

`include.schema.comments`

Whether to include database comments in schemas; may increase memory use.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; may increase memory usage

`column.propagate.source.type`

Adds source database type information for matching columns.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of column regular expressions

`datatype.propagate.source.type`

Adds source type information for columns matching database type patterns.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of database type regular expressions

Snapshots

`snapshot.tables.order.by.row.count`

Orders snapshot tables by row count.

Type: STRING
Default: disabled
Importance: Medium
Valid Values / Notes: ascending, descending, disabled

Heartbeat

`heartbeat.action.query`

Database query executed when sending a heartbeat.

Type: STRING
Default: None
Importance: Low
Valid Values / Notes: executed only when heartbeat interval is non-zero; Depends on: heartbeat.interval.ms

Events

`include.query`

Whether to include original SQL in events; requires MySQL row-query logging and may expose filtered data.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; requires MySQL binlog_rows_query_log_events=ON; may expose excluded data

Filtering

`table.ignore.builtin`

Whether to ignore built-in database tables.

Type: BOOLEAN
Default: true
Importance: Low
Valid Values / Notes: boolean; Depends on: database.include.list

`database.include.list`

Captures only matching databases; the value is a comma-separated list of regular expressions.

Type: LIST
Default: None
Importance: High
Valid Values / Notes: list of regular expressions; mutually exclusive with database.exclude.list; Depends on: database.exclude.list, table.include.list

`database.exclude.list`

Excludes matching databases; cannot be configured together with database.include.list.

Type: LIST
Default: None
Importance: Medium
Valid Values / Notes: list of regular expressions; mutually exclusive with database.include.list; Depends on: database.include.list

Throughput and Buffering

`binlog.buffer.size`

Binlog lookahead buffer size for identifying transaction commits or rollbacks.

Type: INT
Default: 0
Importance: Medium
Valid Values / Notes: non-negative; 0 disables look-ahead buffering

Error Handling

`event.deserialization.failure.handling.mode`

Handling policy for event deserialization failures; deprecated.

Type: STRING
Default: fail
Importance: Medium
Valid Values / Notes: fail, warn, ignore, skip; setting emits a deprecation warning; Deprecated; replacement is event.processing.failure.handling.mode

`inconsistent.schema.handling.mode`

Handling policy for events without a corresponding table schema.

Type: STRING
Default: fail
Importance: Medium
Valid Values / Notes: fail, warn, skip

GTID

`gtid.source.filter.dml.events`

Whether GTID source filtering applies to row-change events.

Type: BOOLEAN
Default: true
Importance: Medium
Valid Values / Notes: boolean; Depends on: gtid.source.includes, gtid.source.excludes

`gtid.source.includes`

Limits the GTID source UUID range; cannot be configured together with gtid.source.excludes.

Type: LIST
Default: None
Importance: High
Valid Values / Notes: GTID source UUID patterns; mutually exclusive with gtid.source.excludes; Depends on: gtid.source.excludes

`gtid.source.excludes`

Excludes the specified GTID source UUIDs; cannot be configured together with gtid.source.includes.

Type: STRING
Default: None
Importance: Medium
Valid Values / Notes: GTID source UUID patterns; rejected when includes is set; Depends on: gtid.source.includes

Metadata

`sourceinfo.struct.maker`

Implementation class producing event Source metadata structures.

Type: CLASS
Default: io.debezium.connector.mysql.MySqlSourceInfoStructMaker
Importance: Low
Valid Values / Notes: SourceInfoStructMaker implementation

Schema History Backend

`schema.history.internal.name`

Logical name for the Schema History backend.

Type: STRING
Default: No declared default; runtime injects <logical-name>-schemahistory
Importance: Low
Valid Values / Notes: backend logical name; Depends on: schema.history.internal

`schema.history.internal.kafka.topic`

The Topic used by Kafka Schema History; it must be a dedicated single-partition Topic that retains the complete DDL history.

Type: STRING
Default: Not declared
Importance: High
Valid Values / Notes: required for KafkaSchemaHistory; Required; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.kafka.bootstrap.servers`

The Kafka cluster address used to read and write Schema History; it should usually be the same cluster used by Connect.

Type: STRING
Default: Not declared
Importance: High
Valid Values / Notes: required for KafkaSchemaHistory; same Kafka cluster is expected; Required; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.kafka.recovery.poll.interval.ms`

Kafka polling interval during Schema History recovery, in milliseconds.

Type: INT
Default: 100
Importance: Low
Valid Values / Notes: non-negative integer; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.kafka.recovery.attempts`

Allowed consecutive empty polls during Schema History recovery.

Type: INT
Default: 100
Importance: Low
Valid Values / Notes: integer; total idle recovery wait is attempts times poll interval; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.kafka.query.timeout.ms`

Timeout for querying Schema History Kafka cluster information, in milliseconds.

Type: LONG
Default: 3000
Importance: Low
Valid Values / Notes: positive integer; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.kafka.create.timeout.ms`

Schema History topic creation timeout in milliseconds.

Type: LONG
Default: 30000
Importance: Low
Valid Values / Notes: positive integer; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.consumer.*`

Client properties passed to the Schema History Kafka consumer.

Type: MAP
Default: Not declared
Importance: Low
Valid Values / Notes: Kafka consumer properties passed through after the prefix; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

`schema.history.internal.producer.*`

Client properties passed to the Schema History Kafka producer.

Type: MAP
Default: Not declared
Importance: Low
Valid Values / Notes: Kafka producer properties passed through after the prefix; Depends on: schema.history.internal=io.debezium.storage.kafka.history.KafkaSchemaHistory

Signal Backend

`signal.file`

File path read by the file signal channel.

Type: STRING
Default: No declared default; runtime falls back to file-signals.txt
Importance: High
Valid Values / Notes: Used only when the file signal channel is enabled; defaults to file-signals.txt when omitted. Depends on: signal.enabled.channels

`signal.kafka.topic`

Topic read by the kafka signal channel.

Type: STRING
Default: No declared default; runtime falls back to <connector logical name>-signal
Importance: High
Valid Values / Notes: Used only when the kafka signal channel is enabled; defaults to <connector logical name>-signal when omitted. Depends on: signal.enabled.channels

`signal.kafka.bootstrap.servers`

Kafka bootstrap servers for the kafka signal channel.

Type: STRING
Default: Not declared
Importance: High
Valid Values / Notes: required when kafka signal channel is enabled; Required; Depends on: signal.enabled.channels

`signal.kafka.poll.timeout.ms`

Poll timeout for the kafka signal channel, in milliseconds.

Type: INT
Default: 0
Importance: Low
Valid Values / Notes: non-negative integer; Depends on: signal.enabled.channels

`signal.kafka.groupId`

Consumer group ID for the kafka signal channel.

Type: STRING
Default: kafka-signal
Importance: Low
Valid Values / Notes: Depends on: signal.enabled.channels

`signal.consumer.*`

Client properties passed to the Kafka signal consumer.

Type: MAP
Default: Not declared
Importance: Low
Valid Values / Notes: Kafka consumer properties passed through after signal.consumer.; Depends on: signal.enabled.channels

Kafka Connect Framework

`name`

The globally unique Connector name.

Type: STRING
Default: Not declared
Importance: High
Valid Values / Notes: non-empty connector name; Required

`connector.class`

The Connector implementation class; MySQL Source uses io.debezium.connector.mysql.MySqlConnector.

Type: STRING
Default: Not declared
Importance: High
Valid Values / Notes: Connector subclass; use io.debezium.connector.mysql.MySqlConnector; Required

`tasks.max`

The maximum number of Tasks this Connector can create; MySQL CDC actually processes only one Task.

Type: INT
Default: 1
Importance: High
Valid Values / Notes: The framework accepts integers of at least 1, but this MySQL connector rejects values greater than 1 with an error when generating task configurations; only 1 is usable.

`tasks.max.enforce`

Whether to enforce tasks.max; disabling it does not bypass the MySQL connector single-task restriction.

Type: BOOLEAN
Default: true
Importance: Low
Valid Values / Notes: true or false; Deprecated; no replacement is defined.

`key.converter`

The Converter class for SourceRecord keys; uses the Worker configuration when not set at the Connector level.

Type: CLASS
Default: None
Importance: Low
Valid Values / Notes: Converter subclass and instantiable

`value.converter`

The Converter class for SourceRecord values; uses the Worker configuration when not set at the Connector level.

Type: CLASS
Default: None
Importance: Low
Valid Values / Notes: Converter subclass and instantiable

`header.converter`

Converter class for SourceRecord headers.

Type: CLASS
Default: None
Importance: Low
Valid Values / Notes: HeaderConverter subclass and instantiable

`transforms`

Single-message transformation aliases in application order.

Type: LIST
Default: empty list
Importance: Low
Valid Values / Notes: unique transformation aliases

`predicates`

Predicate aliases used by single-message transformations.

Type: LIST
Default: empty list
Importance: Low
Valid Values / Notes: unique predicate aliases; Depends on: transforms

`config.action.reload`

Whether to restart the connector when external configuration provider values change.

Type: STRING
Default: restart
Importance: Low
Valid Values / Notes: none, restart

`errors.retry.timeout`

Total framework retry duration for retriable errors, in milliseconds.

Type: LONG
Default: 0
Importance: Medium
Valid Values / Notes: milliseconds; -1 means infinite

`errors.retry.delay.max.ms`

Maximum delay between framework retries, in milliseconds.

Type: LONG
Default: 60000
Importance: Medium
Valid Values / Notes: milliseconds; Depends on: errors.retry.timeout

`errors.tolerance`

Whether framework processing stops or tolerates and skips errors; does not cover arbitrary binlog or startup errors.

Type: STRING
Default: none
Importance: Medium
Valid Values / Notes: none, all

`errors.log.enable`

Whether to log framework processing errors and context.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean

`errors.log.include.messages`

Whether error logs include record contents; enabling this may expose sensitive data.

Type: BOOLEAN
Default: false
Importance: Medium
Valid Values / Notes: boolean; false avoids writing record contents to logs; Depends on: errors.log.enable

Snapshots

`snapshot.locking.mode.custom.name`

SnapshotLock implementation name used by custom locking mode.

Type: STRING
Default: Not declared
Importance: Medium
Valid Values / Notes: SnapshotLock implementation name for snapshot.locking.mode=custom; must match the implementation’s name() result.

`snapshot.select.statement.overrides.<DB>.<TABLE>`

Custom SELECT query for the specified table snapshot.

Type: STRING
Default: Not declared
Importance: Medium
Valid Values / Notes: The table must be listed in snapshot.select.statement.overrides; a missing per-table query produces a warning and that override is skipped.

Schema History Backend

`schema.history.internal.file.filename`

Local history file path for the file-based Schema History backend.

Type: STRING
Default: Not declared
Importance: Medium
Valid Values / Notes: Required when FileSchemaHistory is selected; valid local filesystem path.; Required; Depends on: schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory

Best Practices

Limit the snapshot scope and continue streaming on first connection

Applicable business scenario: Connect to an existing MySQL database for the first time, establish an initial baseline for selected tables only, and continue receiving new changes from those tables after the snapshot completes. Configuration example: Add these settings to Quick Start; remove any conflicting table exclusion list first.

snapshot.mode=initial
snapshot.include.collection.list=<database>\.<table1>,<database>\.<table2>
table.include.list=<database>\.<table1>,<database>\.<table2>

Key points: snapshot.include.collection.list controls which tables are read by this snapshot, while table.include.list also limits the subsequent streaming capture scope. Keep the two lists consistent so that the snapshot baseline and subsequent change scope do not diverge. Keep the Schema History topic dedicated and single-partition, retaining the complete DDL history; do not clean it up like a business topic or arbitrarily delete committed offsets. Normal restarts depend on the original logical identity, offsets, Schema History, and available binlogs.

Capture only changes from Connector startup when existing data is already present

Applicable business scenario: The MySQL database already contains data, but downstream consumers do not need a snapshot of those existing rows and only need INSERT, UPDATE, and DELETE changes captured from Connector startup onward. Configuration example: Add or override the following setting in Quick Start; keep table.include.list as needed to limit the subsequent capture scope.

snapshot.mode=no_data

Key points: no_data does not emit a data snapshot for existing rows, but it still reads table definitions and other required Schema information before continuing with binlog-based CDC. It does not automatically backfill historical changes that occurred before Connector startup; use the initial snapshot scenario or another data initialization process when downstream consumers need an existing-data baseline. Schema History and the required binlogs must still be available, and committed offsets should not be deleted.

Control memory and publishing batches when synchronizing large tables

Applicable business scenario: The ongoing change volume is high or downstream publishing speed fluctuates, requiring a controlled trade-off among memory usage, batch efficiency, and capture latency. Configuration example: Add or override these settings in Quick Start.

max.batch.size=2048
max.queue.size=8192
max.queue.size.in.bytes=67108864
query.fetch.size=1000
poll.interval.ms=500

Key points: The queue absorbs short-term publishing fluctuations; it is not a durable checkpoint. Increasing the queue increases memory usage, and sustained slow publishing still increases capture latency and binlog retention pressure. max.queue.size must be greater than max.batch.size; adjust it gradually based on the Worker heap and downstream throughput.

Monitoring

What to Monitor

Monitor the health and restart counts of the Worker, Connector, and Task; watch Source throughput, end-to-end latency, capture latency, offset commits, errors and retries, queue backlog, and Worker JVM heap and GC. If error tolerance or a dead-letter queue is enabled, also monitor skipped records and DLQ activity; avoid outputting message contents and credentials in error logs.

Import the Grafana Dashboard

Download the JSON from Download the AutoMQ Connect Cluster Dashboard, configure Grafana to use the corresponding metrics data source, ensure the labels match the Connect cluster, and then import the Dashboard in Grafana.

Limitations

MySQL CDC is processed by a single Kafka Connect Task; tasks.max>1 throws an error during task configuration generation and cannot scale CDC throughput in parallel, and snapshot threads do not create additional Tasks.
Include and exclude settings at the same level cannot be used together, such as table.include.list and table.exclude.list, or gtid.source.includes and gtid.source.excludes.
There is no global ordering across Topics for different tables or across Kafka partitions; transaction metadata links transaction information but does not form an atomic downstream transaction.
During failure recovery, events published after the last persisted offset may be duplicated; this Connector does not provide a confirmed end-to-end exactly-once guarantee.
A delete event and a tombstone are separate records; a primary-key change may be sent as a delete for the old key and a create for the new key to different partitions, so consumers must not assume an atomic cross-key update.
Schema History recovery depends on complete history; after binlogs are lost or cleaned up, a new snapshot can recover only current row state and cannot recover each lost intermediate update or delete.

FAQ

What should I do when the Connector reports a MySQL binlog configuration error at startup?

Confirm that MySQL has binary logging enabled and check binlog_format=ROW and binlog_row_image=FULL. Also check the user replication privileges, whether database.server.id is unique, and Connector access to the MySQL port.

Why does increasing `tasks.max` fail to start multiple Tasks?

This is the Connector processing model: MySQL binlog CDC is processed by one Task. tasks.max>1 is rejected when task configurations are generated, rather than ignored; tune filtering scope, batches, queues, and downstream publishing capacity instead.

What should I do when a restart reports missing Schema or an unrecoverable position?

First confirm that the Connector logical identity, offset Topic, Schema History Topic, and Kafka connection configuration have not changed, and that the history Topic was not cleaned up and remains single-partition. Missing Schema History and an expired binlog position are separate problems: when_needed does not handle Schema History errors, and rebuilding current schemas cannot restore missing binlogs or historical schema transitions. A new snapshot establishes current row state, not all intermediate events from expired logs.

Why does the downstream receive a null-valued record after a delete record?

When tombstones.on.delete=true, a same-key null-valued tombstone is sent after the delete event so downstream systems can clean up by key. The delete event itself still contains the change semantics for the deleted row; the two records must not be treated as one record.

Why might an event that was already processed appear again after a restart?

Persisting a Source offset and publishing an event are not one end-to-end atomic transaction. If an event was published before the Worker persisted its corresponding offset and the Worker then failed, recovery may read and publish that event again. Downstream consumers should implement idempotency using the business key and Source metadata, and retain the offset, Schema History, and required binlog together.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Identity

​topic.prefix

​Database Connection

​database.hostname

​database.port

​database.user

​database.password

​database.query.timeout.ms

​database.initial.statements

​database.server.id

​database.server.id.offset

​SSL/TLS

​database.ssl.keystore

​database.ssl.keystore.password

​database.ssl.truststore

​database.ssl.truststore.password

​Database Connection

​database.jdbc.driver

​database.protocol

​SSL/TLS

​database.ssl.mode

​Error Handling

​event.processing.failure.handling.mode

​Throughput and Buffering

​max.batch.size

​max.queue.size

​poll.interval.ms

​max.queue.size.in.bytes

​Events

​provide.transaction.metadata

​skipped.operations

​Snapshots

​snapshot.delay.ms

​streaming.delay.ms

​snapshot.include.collection.list

​snapshot.fetch.size

​snapshot.max.threads

​snapshot.mode.custom.name

​snapshot.mode.configuration.based.snapshot.data

​snapshot.mode.configuration.based.snapshot.schema

​snapshot.mode.configuration.based.start.stream

​snapshot.mode.configuration.based.snapshot.on.schema.error

​snapshot.mode.configuration.based.snapshot.on.data.error

​Error Handling

​retriable.restart.connector.wait.ms

​Throughput and Buffering

​query.fetch.size

​Error Handling

​errors.max.retries

​Incremental Snapshots

​incremental.snapshot.watermarking.strategy

​Database Connection

​connection.validation.timeout.ms

​Runtime

​executor.shutdown.timeout.ms

​Integration

​openlineage.integration.enabled

​openlineage.integration.config.file.path

​openlineage.integration.job.namespace

​openlineage.integration.job.description

​openlineage.integration.job.tags

​openlineage.integration.job.owners

​Events

​extended.headers.enabled

​Type Conversion

​decimal.handling.mode

​Snapshots

​snapshot.lock.timeout.ms

​Database Connection

​connect.timeout.ms

​connect.keep.alive

​connect.keep.alive.interval.ms

​use.nongraceful.disconnect

​Snapshots

​snapshot.mode

Overview

Prerequisites

License

Quick Start

Configuration

Identity

`topic.prefix`

Database Connection

`database.hostname`

`database.port`

`database.user`

`database.password`

`database.query.timeout.ms`

`database.initial.statements`

`database.server.id`

`database.server.id.offset`

SSL/TLS

`database.ssl.keystore`

`database.ssl.keystore.password`

`database.ssl.truststore`

`database.ssl.truststore.password`

Database Connection

`database.jdbc.driver`

`database.protocol`

SSL/TLS

`database.ssl.mode`

Error Handling

`event.processing.failure.handling.mode`

Throughput and Buffering

`max.batch.size`

`max.queue.size`

`poll.interval.ms`

`max.queue.size.in.bytes`

Events

`provide.transaction.metadata`

`skipped.operations`

Snapshots

`snapshot.delay.ms`

`streaming.delay.ms`

`snapshot.include.collection.list`

`snapshot.fetch.size`

`snapshot.max.threads`

`snapshot.mode.custom.name`

`snapshot.mode.configuration.based.snapshot.data`

`snapshot.mode.configuration.based.snapshot.schema`

`snapshot.mode.configuration.based.start.stream`

`snapshot.mode.configuration.based.snapshot.on.schema.error`

`snapshot.mode.configuration.based.snapshot.on.data.error`

Error Handling

`retriable.restart.connector.wait.ms`

Throughput and Buffering

`query.fetch.size`

Error Handling

`errors.max.retries`

Incremental Snapshots

`incremental.snapshot.watermarking.strategy`

Database Connection

`connection.validation.timeout.ms`

Runtime

`executor.shutdown.timeout.ms`

Integration

`openlineage.integration.enabled`

`openlineage.integration.config.file.path`

`openlineage.integration.job.namespace`

`openlineage.integration.job.description`

`openlineage.integration.job.tags`

`openlineage.integration.job.owners`

Events

`extended.headers.enabled`

Type Conversion

`decimal.handling.mode`

Snapshots

`snapshot.lock.timeout.ms`

Database Connection

`connect.timeout.ms`

`connect.keep.alive`

`connect.keep.alive.interval.ms`

`use.nongraceful.disconnect`

Snapshots

`snapshot.mode`