Create a MySQL (Debezium) Source Connector
You can use a MySQL (Debezium) Source connector to import a stream of changes from MySQL, AmazonRDS, and Amazon Aurora.
- A MySQL database that is accessible from the connector instance.
- A MySQL user exists. This database user for the Debezium connector must have LOCK TABLES privileges. For details, see MySQL Creating a user.
- A binlog must be enabled for the source MySQL cluster.
AVROformats can be used as a a Kafka message key and value format.
- The MySQL (Debezium) Source connector can work with only a single task at a time.
Create a MySQL (Debezium) Source connector
To create the MySQL (Debezium) Source connector:
In Redpanda Cloud, click Connectors in the navigation menu, and then click Create Connector.
Select Import from MySQL (Debezium).
On the Create Connector page, specify the following required connector configuration options:
A topic prefix that identifies and provides a namespace for the particular database server/cluster that is capturing changes. The topic prefix should be unique across all other connectors because it is used as a prefix for all Kafka topic names that receive events emitted by this connector. Only alphanumeric characters, hyphens, dots, and underscores are accepted.
A resolvable hostname or IP address of the MySQL database server.
Integer port number of the MySQL database server.
Name of the MySQL user to be used when connecting to the MySQL database.
The password of the MySQL database user who will be connecting to the MySQL database.
The name of the database from which the connector will capture changes.
Specifies whether to use an encrypted connection to the PostgreSQL server. Select
disableto use an unencrypted connection. Select
requireto use a secure, or encrypted connection. If a secure connection cannot be established when
requiredis selected, then the connector fails.
Kafka message key format
Format of the key in the Kafka topic.
Message key JSON contains schema
Enable to specify that the message key contains schema in the schema field.
Kafka message value format
Format of the value in the Kafka topic.
Message value JSON contains schema
Enable to specify that the message value contains schema in the schema field.
Globally-unique name to use for this connector.
Click Next. Review the connector properties specified, then click Create.
Include tables, and
Include columns to define data
mapping. Alternatively, use
Exclude tables, and
Following is an example table in
CREATE TABLE IF NOT EXISTS Persons
Id int PRIMARY KEY,
The table has one record:
INSERT INTO Persons (FirstName, LastName) VALUES (1, 'Winnie', 'the Pooh');
The connector configuration for the table:
column.include.list = db\\.Persons\\.(Id|FirstName|LastName)
table.include.list = db\\.Persons
database.include.list = db
topic.prefix = frommysql
The connector configuration will create the Redpanda topic
Kafka message value format = JSON, the connector produces JSON messages
with a schema like the following:
// schema definition
"LastName": "the Pooh"
Kafka message value format = AVRO, the connector creates a Schema Registry
frommysql.db.Persons-value record and produces messages like the following:
"string": "the Pooh"
Test the connection
After the connector is created:
- Check the connector status and confirm that there are no errors in logs and in Redpanda Console.
- Review the Redpanda topic to confirm that it contains the expected data.
If the connector configuration is invalid, an error appears upon clicking Finish.
Topics not created by the connector
Create the topic manually or let the connector create it by setting (use desired number of partitions and replication factor):
Topic creation enabled: true
Topic creation partitions: 1
Topic creation replication factor: -1
Or in JSON:
Connector requires binlog file 'mysql-bin-changelog.257116', but MySQL only has mysql-bin-changelog.257123
Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted"
Connector requires binlog file 'mysql-bin-changelog.257116', but MySQL only has mysql-bin-changelog.257123, mysql-bin-changelog.257124, mysql-bin-changelog.257125
The connector needs a binlog file that was already purged. Change the
Snapshot modeproperty from the default to
Additional errors and corrective actions follow.
|Unable to connect: Public Key Retrieval is not allowed||Set |
|Unable to connect: Communications link failure||Confirm that |
|Access denied for user||Confirm that |
|Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Invalid schema Invalid namespace: from-mysql.db.Persons; error code: 422||The Schema Registry namespace is incorrect. Consider changing the |