> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Debezium MySQL Source Connector

> Configure and operate the Debezium MySQL Source Connector in AutoMQ Connect, including snapshots, binlog streaming, schema history, security, monitoring, and troubleshooting.

## 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](../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.

```properties theme={null}
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.

```properties theme={null}
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.

```properties theme={null}
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.

```properties theme={null}
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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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.