MySQL Source Connector for Confluent Cloud


If you are installing the connector locally for Confluent Platform, see JDBC Connector (Source and Sink) for Confluent Platform.

The Kafka Connect MySQL Source connector can capture a snapshot of the existing data in a MySQL database and then monitor and record all subsequent row-level changes to that data. All of the events for each table are recorded in a separate Apache Kafka® topic, where they can be easily consumed by applications and services. Note that deleted records are not captured.


The Confluent Cloud MySQL source connector provides the following features:

  • Insert modes:

    • timestamp mode is enabled when only a timestamp column is specified when you enter database details.

    • timestamp+incrementing mode is enabled when both a timestamp column and incrementing column are specified when you enter database details.


      A timestamp column must not be nullable.

  • Database authentication: password authentication.

  • Data formats: Avro, JSON. For JSON, the underlying default configuration property is changed to value.converter.schemas.enable=false.

  • Select configuration properties:

    • db.timezone


    • batch.max.rows


    • topic.prefix

    • schema.pattern

      Configuration properties that are not shown in the Confluent Cloud UI use the default values. See JDBC Source Connector Configuration Properties for default values and property definitions.

Refer to Confluent Cloud connector limitations for additional information.

Quick Start

Use this quick start to get up and running with the Confluent Cloud MySQL source connector. The quick start shows how to select the connector and configure it to capture a snapshot of the existing data in a MySQL database. It then monitors and records all subsequent row-level changes.

  • Authorized access to a Confluent Cloud cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud Platform (GCP).

  • The Confluent Cloud CLI installed and configured for the cluster. See Install and Configure the Confluent Cloud CLI.

  • Public access must be enabled for your database. The example below shows the AWS Management Console when setting up a MySQL database.

    AWS example showing public access for MySQL

    Public access enabled

  • Public inbound traffic access ( must be allowed to the VPC where the database is located. The example below shows the AWS Management Console when setting up security group rules for the VPC.

    AWS example showing security group rules

    Open inbound traffic


    See your specific cloud platform documentation for how to configure security rules for your VPC.

  • At least one topic must exist before creating the connector. If you want a topic prefix, the name of the topic you create must include the prefix.

  • Confluent Cloud Schema Registry must be enabled if you use Avro.

  • Use one of the following for the Kafka cluster credentials fields:

    • A Confluent Cloud API key and secret. After you have created your cluster, go to Cluster settings > API access > Create Key.
    • A Confluent Cloud service account.

Step 1: Launch your Confluent Cloud cluster.

See the Confluent Cloud Quick Start for installation instructions.

Step 2: Add a connector.

Click Connectors > Add connector.


Step 3: Select your connector.

Click the MySQL Source connector icon.

Step 4: Set up the connection.

Complete the following and click Continue.


Make sure you have all your prerequisites completed.

  1. Enter a connector name.

  2. Enter your Kafka Cluster credentials. The credentials are either the API key and secret or the service account API key and secret.

  3. Enter the topic prefix for the database table name. You use this configuration to specify a Kafka topic (or topics), since this connector creates a topic (or topics) directly based on table names from your database.


    You must create the topic names before before creating and launching this connector. To use a topic prefix, you must name the topic using the prefix beforehand. For example, if the database table being sourced is named products and you want the topic to have a prefix list-, create a topic named list-products. You would then enter the prefix list- in the prefix field.

  4. Add the connection details for the database.


    Do not include jdbc:xxxx:// in the Connection host field. The example below shows a sample host address.

  5. Add the Database details for your database. Review the following notes for more information about field selections.

    • Enter a Timestamp column name to enable timesamp mode. This mode uses a timestamp (or timestamp-like) column to detect new and modified rows. This assumes the column is updated with each write, and that values are monotonically incrementing, but not necessarily unique.
    • Enter both a Timestamp column name and an Incrementing column name to enable timestamp+incrementing mode. This mode uses two columns, a timestamp column that detects new and modified rows, and a strictly incrementing column which provides a globally unique ID for updates so each row can be assigned a unique stream offset.
    • By default, the connector only detects tables with type TABLE from the source database. Use VIEW for virtual tables created from joining one or more tables. Use ALIAS for tables with a shortened or temporary name.
  6. Add the Connection details for your connection to the database.

  7. Select how your messages are formatted.

  8. Enter the number of tasks in use by the connector. Refer to Confluent Cloud connector limitations for additional information.

Step 5: Launch the connector.

Verify the connection details and click Launch.


Step 6: Check the connector status.

The status for the connector should go from Provisioning to Running. It may take a few minutes.


Step 7: Check the Kafka topic.

After the connector is running, verify that messages are populating your Kafka topic.

For more information about this connector, see the JDBC Source Connector for Confluent Platform. Note that not all Confluent Platform connector features are provided in the Confluent Cloud connector.

Next Steps

Try out a Confluent Cloud demo.