Skip to main content

Create a Debezium source connector from MySQL to Apache Kafka®

The MySQL Debezium source connector extracts changes committed to the database binary log (binlog) and writes them to an Apache Kafka® topic in a standard format, allowing them to be transformed and read by multiple consumers.

note

Breaking changes in Debezium 2.5

Debezium version 2.5 introduces significant changes to the connector's configuration and behavior. New setups default to version 2.5, while existing setups using version 1.9 remain on that version for stability. Test configurations with version 2.5 before upgrading. For help with upgrades or queries, contact Aiven support.

Schema versioning​

Database table schemas evolve by adding, modifying, or removing columns. The MySQL Debezium source connector tracks schema changes by storing them in a separate history topic, which you can set up with dedicated history.* configuration parameters.

warning

The MySQL Debezium source connector's history.* parameters are not visible in the list of configuration options in the Aiven Console. However, you can insert or modify them by editing the JSON configuration in the Connector configuration section.

Prerequisites​

Before configuring a Debezium source connector for MySQL, ensure you have:

Required database details​

Gather the following details about your MySQL database:

note

You can view the full set of available parameters and configuration options in the connector's documentation.

  • MYSQL_HOST: The database hostname.
  • MYSQL_PORT: The database port.
  • MYSQL_USER: The database user to connect.
  • MYSQL_PASSWORD: The database password for the MYSQL_USER.
  • MYSQL_DATABASE_NAME: The database name.
  • SSL_MODE: The SSL mode.
  • MYSQL_TABLES: The list of database tables to be included in Apache Kafka. Format the list as schema_name1.table_name1,schema_name2.table_name2.
  • APACHE_KAFKA_HOST: The hostname of the Apache Kafka service, needed when storing the schema definition changes.
  • APACHE_KAFKA_PORT: The port of the Apache Kafka service, needed when storing the schema definition changes.
  • SCHEMA_REGISTRY_PORT: The Apache Kafka's schema registry port is only needed when using Avro as a data format.
  • SCHEMA_REGISTRY_USER: The Apache Kafka's schema registry username is only needed when using Avro as a data format.
  • SCHEMA_REGISTRY_PASSWORD: The Apache Kafka's schema registry user password is only needed when using Avro as a data format.
note

If you're using Aiven for MySQL and Aiven for Apache Kafka, find the required details on the service's Overview page in the Aiven console or by running the avn service get command via the Aiven CLI.

Create an Apache Kafka Connect configuration file​

Create a file named debezium_source_mysql.json with the following connector configurations. This file is optional but helps you keep your settings organized and makes it easier to copy and paste them into the Aiven Console later.

{
    "name":"CONNECTOR_NAME",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "MYSQL_HOST",
    "database.port": "MYSQL_PORT",
    "database.user": "MYSQL_USER",
    "database.password": "MYSQL_PASSWORD",
    "database.dbname": "MYSQL_DATABASE_NAME",
    "database.ssl.mode": "SSL_MODE",
    "database.server.id": "UNIQUE_ID",
    "topic.prefix": "KAFKA_TOPIC_PREFIX",
    "table.include.list": "MYSQL_TABLES",
    "tasks.max":"NR_TASKS",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT",
    "key.converter.basic.auth.credentials.source": "USER_INFO",
    "key.converter.schema.registry.basic.auth.user.info": "SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT",
    "value.converter.basic.auth.credentials.source": "USER_INFO",
    "value.converter.schema.registry.basic.auth.user.info": "SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD",
    "schema.history.internal.kafka.topic": "HISTORY_TOPIC_NAME",
    "schema.history.internal.kafka.bootstrap.servers": "APACHE_KAFKA_HOST:APACHE_KAFKA_PORT",
    "schema.history.internal.producer.security.protocol": "SSL",
    "schema.history.internal.producer.ssl.keystore.type": "PKCS12",
    "schema.history.internal.producer.ssl.keystore.location": "/run/aiven/keys/public.keystore.p12",
    "schema.history.internal.producer.ssl.keystore.password": "password",
    "schema.history.internal.producer.ssl.truststore.location": "/run/aiven/keys/public.truststore.jks",
    "schema.history.internal.producer.ssl.truststore.password": "password",
    "schema.history.internal.producer.ssl.key.password": "password",
    "schema.history.internal.consumer.security.protocol": "SSL",
    "schema.history.internal.consumer.ssl.keystore.type": "PKCS12",
    "schema.history.internal.consumer.ssl.keystore.location": "/run/aiven/keys/public.keystore.p12",
    "schema.history.internal.consumer.ssl.keystore.password": "password",
    "schema.history.internal.consumer.ssl.truststore.location": "/run/aiven/keys/public.truststore.jks",
    "schema.history.internal.consumer.ssl.truststore.password": "password",
    "schema.history.internal.consumer.ssl.key.password": "password",
    "include.schema.changes": "true"
}

Parameters:

  • name: The connector name. Replace CONNECTOR_NAME with the connector name.

  • database.hostname, database.port, database.dbname, database.ssl.mode, database.user, database.password, table.include.list: Source database parameters collected in the prerequisite phase.

  • database.server.id: The logical name of the database server. This should be a unique ID.

  • topic.prefix: The prefix to be used for the Apache Kafka topic names.

  • tasks.max: Maximum number of tasks to execute in parallel. Replace NR_TASKS with the number of parallel tasks based on the number of tables.

  • schema.history.internal.kafka.topic: The name of the Apache Kafka topic that contains the history of schema changes.

  • schema.history.internal.kafka.bootstrap.servers: The address of the Apache Kafka service that stores schema definition changes.

  • schema.history.internal.producer and schema.history.internal.consumer: Parameters related to SSL configuration for Apache Kafka connections.

    warning

    The values defined for each schema.history.internal.producer and schema.history.internal.consumer parameters are already set to work with the predefined truststore and keystore created in the Aiven for Apache Kafka nodes. Modifying these values is not recommended.

  • key.converter and value.converter: Defines the message data format in the Apache Kafka topic. Avro format is used in this example.

  • include.schema.changes: Whether to include changes to the schema itself.

    note

    The key.converter and value.converter sections are only needed when pushing data in Avro format. Otherwise, messages default to JSON format.

    The USER_INFO is not a placeholder and does not require any parameter substitution.

Setup a MySQL Debezium source connector​

  1. Access the Aiven Console.

  2. Select your Aiven for Apache Kafka® or Aiven for Apache Kafka Connect® service.

  3. Click Connectors.

  4. Click Create connector if Apache Kafka Connect is already enabled on the service. If not, click Enable connector on this service.

    Alternatively, to enable connectors:

    1. Click Service settings in the sidebar.
    2. In the Service management section, click Actions > Enable Kafka connect.

  5. In the source connectors, click Get started on Debezium - MySQL.

  6. On the Create Debezium - MySQL Connector page, click the Common tab, find the Connector configuration text box, and click Edit.

  7. Paste the connector configuration stored in the debezium_source_mysql.json file into the editor.

  8. Click Apply.

    note

    The Aiven Console reads the configuration file and automatically populates the relevant UI fields. You can view and modify these fields across different tabs. Any changes you make in these fields are reflected in the JSON format within the Connector configuration text box.

  9. Click Create connector.

    tip

    With Aiven for Apache Kafka, topics are not created automatically. You have two options:

    • Manually create topics using the naming pattern: database.server.name.schema_name.table_name.
    • Enable auto-creation for Apache Kafka topics.

  10. Verify the connector status on the Connectors page.

Verify that data from the MySQL dataset is present in the target Apache Kafka topic. The topic name is a concatenation of the database and table name. To modify the target table name, use the Apache Kafka Connect RegexRouter transformation.