# MySQL CDC Source V2 (Debezium) Connector for Confluent Cloud The fully managed MySQL Change Data Capture (CDC) Source V2 (Debezium) connector for Confluent Cloud streams row-level changes from a MySQL database into Apache Kafka® topics, using the Debezium engine internally. The connector can also take an initial snapshot of existing data before streaming subsequent `INSERT`, `UPDATE`, and `DELETE` changes. Each table’s events are recorded to a separate Kafka topic, and the connector supports Avro, JSON Schema, Protobuf, or JSON (schemaless) output formats. #### NOTE * This Quick Start is for version 2 of fully managed Confluent Cloud connector. For the earlier version of this connector, see [MySQL CDC Source (Debezium) Connector [End of Life] for Confluent Cloud](../cc-mysql-source-cdc-debezium.md#cc-mysql-source-cdc-debezium). If moving from V1 to V2, see [Moving from V1 to V2](#cc-mysql-cdc-source-v2-debezium-migrating). * If you are installing the connector locally for Confluent Platform, see [Debezium MySQL CDC Source Connector Connector for Confluent Platform](https://docs.confluent.io/kafka-connectors/debezium-mysql-source/current/). * MariaDB is now supported. For more information on connecting to MariaDB, see [MariaDB CDC Source connector](../cc-mariadb-cdc-source-debezium.md#cc-mariadb-source-cdc-debezium-configure-connector) documentation. * If you require private networking for fully managed connectors, make sure to set up the proper networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud). ## V2 Improvements Note the following improvements made to the V2 connector. * Supports reading binlog entries that were written with compression enabled. * Supports parsing JSON_TABLE table functions. * Can stop or pause an in-progress incremental snapshot. Can resume the incremental snapshot if it was previously been paused. * Supports regular expressions to specify table names for incremental snapshots. * Supports SQL-based predicates to control the subset of records to be included in the incremental snapshot. * Supports specifying a single column as a surrogate key for performing incremental snapshots. * Supports the `additional-condition` option of the [signaling](https://debezium.io/documentation/reference/1.9/configuration/signalling.html#sending-signals-to-a-debezium-connector) feature for incremental snapshots. * Can perform ad-hoc blocking snapshots. * Indices that rely on hidden, auto-generated columns, or columns wrapped in database functions are no longer considered primary key alternatives for tables that do not have a primary key defined. * Configuration options to specify how topic and schema names should be adjusted for compatibility. ## Features The MySQL CDC Source V2 (Debezium) connector provides the following features: * **Topics created automatically**: The connector automatically creates Kafka topics using the naming convention: `..`. The tables are created with the properties: `topic.creation.default.partitions=1` and `topic.creation.default.replication.factor=3`. For more information, see [Maximum message size](#cc-mysql-cdc-source-v2-debezium-topic-sizing). * **Database authentication:** Uses password authentication or AWS IAM role-based authentication via Confluent Provider Integration. * **Provider integration support**: The connector supports AWS IAM role-based authorization using Confluent Provider Integration. For more information about provider integration setup, see [IAM roles authentication](#cc-mysql-cdc-source-v2-debezium-setup-connection) and [AWS IAM Authentication Setup](#cc-mysql-cdc-source-v2-debezium-aws-iam-auth). * **SSL support**: Supports SSL encryption. * **Databases included** and **Databases excluded**: Sets whether a database is or is not monitored for changes. By default, the connector monitors every database on the server. * **Tables included** and **Tables excluded**: Sets whether a table is or is not monitored for changes. By default, the connector monitors every non-system table. * **Tombstones on delete**: Sets whether a tombstone event is generated after a delete event. Default is `true`. * **Output formats:** The connector supports Avro, JSON Schema, Protobuf, or JSON (schemaless) *output Kafka record value format*. It supports Avro, JSON Schema, Protobuf, JSON (schemaless), and String *output record key format*. [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). * **Incremental snapshot**: Supports incremental snapshotting via [signaling](https://debezium.io/documentation/reference/1.9/configuration/signalling.html#sending-signals-to-a-debezium-connector). * **Offset management capabilities**: Supports offset management. For more information, see [Manage custom offsets](#cc-mysql-cdc-source-v2-debezium-custom-offsets). * **Client-side encryption (CSFLE and CSPE) support**: The connector supports CSFLE and CSPE for sensitive data. For more information about CSFLE or CSPE setup, see the [Manage CSFLE or CSPE for connectors](../csfle.md#connect-csfle). * **Secret manager integration**: The connector supports secret manager integration. For `Password` based authentication, the connector can retrieve the following configurations from an integrated secret manager at runtime as needed. | **Secret manager managed configuration** | **Type** | |--------------------------------------------|------------| | `database.user` | `STRING` | | `database.password` | `PASSWORD` | For more information, see [Create a secret manager integration in Confluent Cloud](../secret-manager-integration/overview.md#cloud-secret-manager-quickstart). For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section. ## Supported database versions The MySQL CDC Source V2 (Debezium) connector is compatible with the following MySQL versions: 8.0.x, 8.4.x. ## Limitations Be sure to review the following information. * For connector limitations, see [MySQL CDC Source V2 (Debezium) Connector](../limits.md#cc-mysql-cdc-source-v2-limits) limitations. * If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](../single-message-transforms.md#cc-single-message-transforms-limitations). - If you plan to use one or more Custom SMTs, see [Custom SMT limitations](../configure-custom-single-message-transforms/custom-smt-limitations-support.md#cc-custom-smt-limitations). ## Deprecated features and configurations The following features and configuration properties have been deprecated. Confluent recommends using the alternatives instead: * **Snapshot mode**: The `schema_only` snapshot mode is deprecated. Use `no_data` instead. * **Signaling**: The `additional-condition` option in the signal query is deprecated. Use `additional-conditions` instead. * **ExtractNewRecordState SMT**: The configurations `drop.tombstones` and `delete.handling.mode` configurations are deprecated. Use `delete.tombstones.handling.mode` instead. ## Maximum message size This connector creates topics automatically. When it creates topics, the internal connector configuration property `max.message.bytes` is set to the following: * Basic cluster: `8 MB` * Standard cluster: `8 MB` * Enterprise cluster: `8 MB` * Dedicated cluster: `20 MB` For more information about Confluent Cloud clusters, see [Kafka Cluster Types in Confluent Cloud](../../clusters/cluster-types.md#cloud-cluster-types). ## Log retention during snapshot When launched, the CDC connector creates a snapshot of the existing data in the database to capture the nominated tables. To do this, the connector executes a `“SELECT *”` statement. Completing the snapshot can take a while if one or more of the nominated tables is very large. During the snapshot process, the database server must retain transaction logs so that when the snapshot is complete, the CDC connector can start processing database changes that have completed since the snapshot process began. These logs are retained in a binary log (binlog) on the database server. **Binlog retention recommendation** Confluent recommends setting the binlog retention period to **7 days** (for example, `expire_logs_days = 7`). This ensures that the connector has access to older binlog files in case of downtime or other incidents. Resuming after a downtime, the MySQL connector might need to access older binlog files to resume processing changes from last committed offset. **Risks of low binlog retention** If the binlog retention period is set too low, the following problems can occur: * **Connector failure after downtime**: If the connector is stopped for longer than the binlog retention period, the required binlog files may be purged. This can cause the connector to fail or require a full snapshot to recover. * **Snapshot failures**: For very large tables, the snapshot process might exceed the binlog retention time. When the snapshot completes, the connector may not be able to find the binlog files needed to process changes that occurred during the snapshot, leading to data inconsistency or connector failures. * **Data loss**: If binlog files are purged before the connector has processed them, changes to the database may not be captured. This results in missing data in the target topics. ## Manage custom offsets You can manage the offsets for this connector. Offsets provide information on the point in the system from which the connector is accessing data. For more information, see [Manage Offsets for Fully Managed Connectors in Confluent Cloud](../offsets.md#connect-custom-offsets). **To manage offsets**: - Manage offsets using Confluent Cloud APIs. For more information, see [Connect offsets API reference](https://docs.confluent.io/cloud/current/ccloud/offsets-connect-v-1/). ### Get the current offset To get the current offset, make a `GET` request that specifies the environment, Kafka cluster, and connector name. ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets Host: https://api.confluent.cloud ``` **Response:** Successful calls return HTTP `200` with a JSON payload that describes the offset. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": { "server": "server_01" }, "offset": { "event": 2, "file": "mysql-bin.000598", "pos": 2326, "row": 1, "server_id": 1, "transaction_id": null, "ts_sec": 1711648627 } } ], "metadata": { "observed_at": "2024-03-28T17:57:48.139635200Z" } } ``` Responses include the following information: - The position of latest offset. - The observed time of the offset in the metadata portion of the payload. The `observed_at` time indicates a snapshot in time for when the API retrieved the offset. A running connector is always updating its offsets. Use `observed_at` to get a sense for the gap between real time and the time at which the request was made. By default, offsets are observed every minute. Calling get repeatedly will fetch more recently observed offsets. - Information about the connector. ### Update the offset To update the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector name. Include a JSON payload that specifies new offset and a patch type. ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request Host: https://api.confluent.cloud { "type": "PATCH", "offsets": [ { "partition": { "server": "server_01" }, "offset": { "event": 2, "file": "mysql-bin.000598", "pos": 1423, "row": 1, "server_id": 1, "transaction_id": null, "ts_sec": 1711648518 } } ] } ``` Considerations: - You can only make one offset change at a time for a given connector. - This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information, see **Get the status of an offset request**. - For source connectors, the connector attempts to read from the position defined by the requested offsets. **Response:** Successful calls return HTTP `202 Accepted` with a JSON payload that describes the offset. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": { "server": "server_01" }, "offset": { "event": 2, "file": "mysql-bin.000598", "pos": 1423, "row": 1, "server_id": 1, "transaction_id": null, "ts_sec": 1711648518 } } ], "requested_at": "2024-03-28T17:58:45.606796307Z", "type": "PATCH" } ``` Responses include the following information: - The requested position of the offsets in the source. - The time of the request to update the offset. - Information about the connector. ### Delete the offset To delete the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector name. Include a JSON payload that specifies the delete type. ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request Host: https://api.confluent.cloud { "type": "DELETE" } ``` Considerations: - Delete requests delete the offset for the provided partition and reset to the base state. A delete request is as if you created a fresh new connector. - This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information, see **Get the status of an offset request**. - Do not issue delete and patch requests at the same time. - For source connectors, the connector attempts to read from the position defined in the base state. **Response**: Successful calls return HTTP `202 Accepted` with a JSON payload that describes the result. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [], "requested_at": "2024-03-28T17:59:45.606796307Z", "type": "DELETE" } ``` Responses include the following information: - Empty offsets. - The time of the request to delete the offset. - Information about Kafka cluster and connector. - The type of request. ### Get the status of an offset request To get the status of a previous offset request, make a `GET` request that specifies the environment, Kafka cluster, and connector name. ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request/status Host: https://api.confluent.cloud ``` Considerations: - The status endpoint always shows the status of the most recent PATCH/DELETE operation. **Response**: Successful calls return HTTP `200` with a JSON payload that describes the result. The following is an example of an applied patch. ```bash { "request": { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": { "server": "server_01" }, "offset": { "event": 2, "file": "mysql-bin.000598", "pos": 1423, "row": 1, "server_id": 1, "transaction_id": null, "ts_sec": 1711648518 } } ], "requested_at": "2024-03-28T17:58:45.606796307Z", "type": "PATCH" }, "status": { "phase": "APPLIED", "message": "The Connect framework-managed offsets for this connector have been altered successfully. However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses." }, "previous_offsets": [ { "partition": { "server": "server_01" }, "offset": { "event": 2, "file": "mysql-bin.000598", "pos": 2326, "row": 1, "server_id": 1, "transaction_id": null, "ts_sec": 1711648627 } } ], "applied_at": "2024-03-28T17:58:48.079141883Z" } ``` Responses include the following information: - The original request, including the time it was made. - The status of the request: applied, pending, or failed. - The time you issued the status request. - The previous offsets. These are the offsets that the connector last updated prior to updating the offsets. Use these to try to restore the state of your connector if a patch update causes your connector to fail or to return a connector to its previous state after rolling back. ### JSON payload The table below offers a description of the unique fields in the JSON payload for managing offsets of the MySQL Change Data Capture (CDC) Source connector. | Field | Definition | Required/Optional | |------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------| | `event` | The number of rows and events to skip while starting from this file and position. Use `event` and `row` together
but only use these fields if you understand which rows and events to skip. For most cases, you only need to
provide `file` and `position`. | Optional | | `file` | The file from the last processed binlog. Use `file` and `pos` together. | Required | | `pos` | The position from the last processed binlog. Use `pos` and `file` together. | Required | | `row` | The number of rows and events to skip while starting from this file and position. Use `event` and `row` together
but only use these fields if you understand which rows and events to skip. For most cases, you only need to
provide `file` and `position`. | Optional | | `server_id` | The id of the server from which the event originated. For more information, see [MySQL documentation](https://dev.mysql.com/doc/refman/8.0/en/replication-options.html#sysvar_server_id). | Optional | | `transaction_id` | Mostly null, provided only when `provide.transaction.metadata` is set to `true` in the config. For more information, see [Debezium Documentation](https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-property-provide-transaction-metadata). | Optional | | `ts_sec` | The timestamp at which the event at this pos was executed in the database. | Optional | #### IMPORTANT Do not reset the offset to an arbitrary number. Use only offsets found in the binlog file. To find offsets in a binlog file, use the `mysqlbinlog` utility. Offsets appear in this format: `# at ` ### Migrate connectors Considerations: - The configurations of the self-managed connector must match the configurations of the fully managed connector. - The self-managed connector must be operating in streaming mode. If the self-managed connector is still in the process of making a snapshot, you can either create a new connector on Confluent Cloud which starts the snapshot process from the beginning or wait for the snapshot process to complete and follow the migration guidance. ### Create fully managed connectors with offsets Considerations: - Schema history topic of the self-managed connector must be reused while creating the fully managed connector. This can be done by specifying the `Database schema history topic name` config value in the Confluent Cloud Console or `schema.history.internal.kafka.topic` config value using Confluent CLI and snapshot mode set to `no_data`. - To reuse schema history topic, the specified value of `Topic prefix` should be same as that of the self-managed connector. - If schema history topic of the self-managed connector is not available or can not be reused, you can start the connector with `recovery` snapshot mode. This populates the schema history topic first and then starts the connector from the specified offset. The connector will fail if there have been schema changes in the included tables after the specified offset. ## Quick Start Use this quick start to get up and running with the MySQL CDC Source V2 (Debezium) connector. The quick start provides the basics of selecting the connector and configuring it to obtain a snapshot of the existing data in a MySQL database and then monitoring and recording all subsequent row-level changes. Prerequisites : - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud. - The Confluent CLI installed and configured for the cluster. See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). - [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). - Public access may be required for your database. See [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud) for details. The example below shows the AWS Management Console when setting up a MySQL database. ![AWS example showing public access for MySQL](images/ccloud-postgresql-source-vpc-setting.png) - For networking considerations, see [Networking and DNS](../overview.md#connect-internet-access-resources). To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](../static-egress-ip.md#cc-static-egress-ips). The example below shows the AWS Management Console when setting up security group rules for the VPC. ![AWS example showing security group rules](images/ccloud-postgresql-source-inbound-setting.png)
#### NOTE See your specific cloud platform documentation for how to configure security rules for your VPC.
- Kafka cluster credentials. The following lists the different ways you can provide credentials. - Enter an existing [service account](../service-account.md#s3-cloud-service-account) resource ID. - Create a Confluent Cloud [service account](../service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](../service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements. - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.
- Update the following settings for the MySQL database. 1. Turn on backup for the database. 2. Create a new parameter group and set the following parameters: ```none binlog_format=ROW binlog_row_image=full expire_logs_days=7 ``` 3. Apply the new parameter group to the database. 4. Reboot the database.
#### NOTE Set the `expire_logs_days` to `7`. This ensures the connector can access old binlog files in case of downtime or other incidents. For more information about binlog retention, see [Log retention during snapshot](#cc-mysql-cdc-source-v2-debezium-log-retention).
The following example screens are from Amazon RDS: ![Set database backup](images/ccloud-mysql-cdc-source-debezium-set-backup.png)![Set database binlog](images/ccloud-mysql-cdc-source-debezium-set-binlog.png)![Set database binlog row image](images/ccloud-mysql-cdc-source-debezium-set-binlog_row.png) ### Using the Confluent Cloud Console #### Step 1: Launch your Confluent Cloud cluster To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../../get-started/index.md#cloud-create-kafka-cluster). #### Step 2: Add a connector In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add connector**. #### Step 3: Select your connector Click the **MySQL CDC Source V2** connector card. ![MySQL CDC Source V2 Connector Card](images/ccloud-mysql-cdc-source-v2-debezium-icon.png) #### Step 4: Enter the connector details #### NOTE * Make sure you have all your [prerequisites](#cc-mysql-cdc-source-v2-debezium-prereqs) completed. At the **MySQL CDC Source V2 (Debezium) Connector** screen, complete the following: ### Kafka access 1. Select the way you want to provide **Kafka Cluster credentials**. You can choose one of the following options: - **My account**: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production. - **Service account**: This setting limits the access for your connector by using a [service account](../service-account.md#s3-cloud-service-account). This option is recommended for production. - **Use an existing API key**: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments. #### NOTE Freight clusters support only service accounts for Kafka authentication. 2. Click **Continue**. ### Authentication 1. Configure the authentication properties: **Authentication method** - **Authentication method**: Select how you want to authenticate with the database. Valid options are `Password` and `AWS IAM Roles`. - **Use secret manager**: Enable this setting to fetch sensitive configuration values, such as `Password`, from a secret manager. - **Provider Integration**: Select an existing integration that has access to your resource such as the secret manager. **Secret manager configuration** - **Secret manager**: Select the secret manager that Confluent Cloud should use to retrieve sensitive data. - **Configurations from Secret manager**: Select the configurations whose values Confluent Cloud should fetch from the secret manager. - **Provider Integration**: Select an existing integration that has access to your resource such as the secret manager. **How should we connect to your database?** - **Database hostname**: IP address or hostname of the MySQL database server. - **Database port**: Port number of the MySQL database server. - **Database username**: The name of the MySQL database user that has the required authorization. - **Database password**: The password for the MySQL database user that has the required authorization. Only required if Auth Mode is Password. - **Database AWS region**: The AWS region of the MySQL database server for RDS or Aurora. - **SSL mode**: Whether to use an encrypted connection to the MySQL server. Valid options are `disabled`, `preferred`, `required`, `verify_ca`, and `verify_identity`. - `preferred` (default): Establishes an encrypted connection if the server supports secure connections. If the server does not support secure connections, the connector falls back to an unencrypted connection. - `disabled`: Uses an unencrypted connection. - `required`: Establishes an encrypted connection or fails if one cannot be made for any reason. - `verify_ca`: Behaves like the `required` option, but also verifies the server TLS certificate against the configured Certificate Authority (CA) certificate. If the server TLS certificate does not match any valid CA certificates, the connector fails. - `verify_identity`: Behaves like the `verify_ca` option, but also verifies that the server certificate matches the host of the remote connection. - **SSL Keystore**: Path to the SSL keystore file for MySQL connection. Only needed when SSL certificate verification is required (verify_ca or verify_identity modes). - **SSL Keystore Password**: Password for the SSL keystore file for MySQL connection. Only needed when SSL certificate verification is required and the keystore is password-protected. - **SSL Truststore**: Path to the SSL truststore file for MySQL connection. Only needed when SSL certificate verification is required (verify_ca or verify_identity modes). - **SSL Truststore Password**: Password for the SSL truststore file for MySQL connection. Only needed when SSL certificate verification is required and the truststore is password-protected. 2. Click **Continue**. ### Configuration **Output messages** - **Select output record value format**: Select the output record value format (data going to the Kafka topic). Valid entries are AVRO, JSON, JSON_SR (JSON Schema), or PROTOBUF. [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON Schema, or Protobuf). - **Output Kafka record key format**: Sets the output Kafka record key format. Valid entries are AVRO, JSON_SR, PROTOBUF, STRING or JSON. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF. **Connector config** - **Snapshot mode**: Specifies the criteria for running a snapshot when the connector starts. Possible settings are: `initial`, `initial_only`, `when_needed`, `never`, `schema_only`, `no_data`, `schema_only_recovery` and `recovery`. - `initial` (default): the connector runs a snapshot only when no offsets have been recorded for the logical server name. - `initial_only`: the connector runs a snapshot only when no offsets have been recorded for the logical server name and then stops. That is, it does not read change events from the binlog. - `when_needed`: the connector runs a snapshot upon startup whenever it deems it necessary. That is, when no offsets are available, or when a previously recorded offset specifies a binlog location or global transaction identifier (GTID) that is not available in the server. Note that if the connector cannot find the binlog file mentioned in the offsets, it will take another snapshot and that may lead to duplicate data. - `never`: the connector never uses snapshots. Upon first startup with a logical server name, the connector reads from the beginning of the binlog. Configure this behavior with care. It is valid only when the binlog is guaranteed to contain the entire history of the database. - `schema_only`: Deprecated, use `no_data` instead. - `no_data`: the connector runs a snapshot of the schemas and not the data. This setting is useful when you do not need the topics to contain a consistent snapshot of the data but need them to have only the changes since the connector was started. - `schema_only_recovery`: Deprecated, use `recovery` instead. - `recovery`: Recovery setting for a connector that has already been capturing changes. When you restart the connector, this setting enables recovery of a corrupted or lost database schema history topic. You might set it periodically to **clean up** a database schema history topic that has been growing unexpectedly. Database schema history topics require infinite retention. - **Databases included**: An optional, comma-separated list of regular expressions that match the names of the databases for which to capture changes. The connector does not capture changes in any database whose name is not in this list. By default, the connector captures changes in all databases. To match the name of a database, Debezium applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire name string of the database; it does not match substrings that might be present in a database name. - **Exclude Databases**: A comma-separated list of regular expressions that match the names of databases from which you do not want the connector to capture changes. The connector captures changes in any database that is not named in the `database.exclude.list` - **Tables included**: Enter a comma-separated list of fully-qualified table identifiers for the connector to monitor. By default, the connector monitors all non-system tables. A fully-qualified table name is in the form `databaseName.tableName`. This can’t be used with `Tables excluded`. - **Tables excluded**: Enter a comma-separated list of fully-qualified table identifiers for the connector to *ignore*. A fully-qualified table name is in the form `databaseName.tableName`. This property cannot be used with the property `Tables included`. **Data encryption** - Enable **Client-Side Field Level Encryption** for data encryption. Specify a **Service Account** to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](../csfle.md#connect-csfle). ### **Show advanced configurations** - **Schema context**: Select a schema context to use for this connector, if using a schema-based data format. This property defaults to the **Default** context, which configures the connector to use the default schema set up for Schema Registry in your Confluent Cloud environment. A schema context allows you to use separate schemas (like schema sub-registries) tied to topics in different Kafka clusters that share the same Schema Registry environment. For example, if you select a non-default context, a **Source** connector uses only that schema context to register a schema and a **Sink** connector uses only that schema context to read from. For more information about setting up a schema context, see [What are schema contexts and when should you use them?](../../sr/faqs-cc.md#faq-schema-contexts). **Additional Configs** - **Value Converter Replace Null With Default**: Specifies whether to replace fields that have a default value and that are null to the default value. When set to `true`, the connector uses the default value; otherwise, it uses `null`. Applies to the `JSON` converter. - **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`. - **Value Converter Schemas Enable**: Includes schema within each of the serialized values. Input messages must contain `schema` and `payload` fields and must not contain additional fields. For plain `JSON` data, set this to `false`. Applies to the `JSON` converter. - **Errors Tolerance**: Use this property to configure the connector’s error handling behavior. #### WARNING Use this property with caution for sink connectors, as it can lead to data loss. If you set this property to `all`, the connector does not fail on errant records, but logs them (and sends to DLQ for sink connectors) and continues processing. If you set this property to `none`, the connector task fails on errant records. - **Value Converter Ignore Default For Nullables**: When set to `true`, this property ensures that the corresponding record in Kafka is `null`, instead of showing the default column value. Applies to the `AVRO`, `PROTOBUF`, and `JSON_SR` converters. - **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals: `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value. - **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers. - **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters. - **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry. - **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration. - **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers. **Auto-restart policy** - **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to `true`, enabling the connector to automatically restart in case of user-actionable errors. Set this property to `false` to disable auto-restart for failed connectors. If disabled, you must manually restart the connector. **Output messages** - **After-state only**: Controls whether the generated Kafka record should contain only the state of the row after the event occurred. Defaults to `false`. - **Tombstones on delete**: Configure whether a tombstone event should be generated after a delete event. The default is `true`. - **Replace null with default**: Applies only when `After-state only` is enabled. When set to `true`, `null` field values are replaced by the source-defined default values when the connector rewrites the record. Set to `false` to preserve `null` field values. For example, clearing a column value or removing a column default is reflected as `null` in the output record. **Database config** - **Signal data collection**: Fully-qualified name of the data collection that is used to send signals to the connector. Use the following format to specify the fully-qualified collection name: `databaseName.tableName`. These signals can be used to perform [incremental snapshotting](https://debezium.io/documentation/reference/2.4/connectors/mysql.html#debezium-mysql-incremental-snapshots). **How should we name your topic(s)?** - **Database schema history topic name**: The name of the topic for the database schema history. A new topic with the provided name is created if it doesn’t already exist. If the topic already exists, ensure that it has a single partition, infinite retention period and is not in use by any other connector. If no value is provided, the name defaults to `dbhistory..`. **Connector config** - **Snapshot locking mode**: Controls whether and how long the connector holds the global MySQL read lock, which prevents any updates to the database, while the connector is performing a snapshot. Possible settings are: `minimal`, `minimal_percona`, `extended`, and `none`. - `minimal`: the connector holds the global read lock for just the initial portion of the snapshot, while the database schemas and other metadata are being read. The remaining work in a snapshot involves selecting all rows from each table. This is accomplished using a REPEATABLE READ transaction, even when the lock is no longer held and other MySQL clients are updating the database. - `minimal_percona`: similar to `minimal` mode except the connector uses a (Percona-specific) backup lock. This mode does not flush tables to disk, is not blocked by long-running reads, and is available only in Percona Server. - `extended`: blocks all writes for the duration of the snapshot. Use this setting if there are clients that are submitting operations that MySQL excludes from REPEATABLE READ semantics. - `none`: prevents the connector from acquiring any table locks during the snapshot. While this setting is allowed with all snapshot modes, it is safe to use if and only if no schema changes are happening while the snapshot is running. For tables defined with MyISAM engine, the tables would still be locked despite this property being set as MyISAM acquires a table lock. This behavior is unlike InnoDB engine, which acquires row level locks. - **Columns excluded**: An optional, comma-separated list of regular expressions that match the fully-qualified names of columns to exclude from change event record values. Fully-qualified names for columns are of the form `databaseName.tableName.columnName`. - **Event processing failure handling mode**: Specifies how the connector should react to exceptions during processing of events. Possible settings are: `fail`, `skip`, and `warn`. - `fail` (default): propagates the exception, indicates the offset of the problematic event, and causes the connector to stop. - `skip`: skips the problematic event and continues processing. - `warn`: logs the offset of the problematic event, skips that event, and continues processing. - **Schema name adjustment mode**: Specifies how schema names should be adjusted for compatibility with the message converter used by the connector. Possible settings are: `none`, `avro`, and `avro_unicode`. - `none` (default): does not apply any adjustment. - `avro`: replaces the characters that cannot be used in the Avro type name with underscore. - `avro_unicode`: replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like \_uxxxx. Note: \_ is an escape sequence like backslash in Java. - **Field name adjustment mode**: Specifies how field names should be adjusted for compatibility with the message converter used by the connector. Possible settings are: `none`, `avro`, and `avro_unicode`. - `none` (default): does not apply any adjustment. - `avro`: replaces the characters that cannot be used in the Avro type name with underscore. - `avro_unicode`: replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like \_uxxxx. Note: \_ is an escape sequence like backslash in Java. - **Heartbeat interval (ms)**: Controls how frequently the connector sends heartbeat messages to a Kafka topic. The behavior of default value 0 is that the connector does not send heartbeat messages. Heartbeat messages are useful for monitoring whether the connector is receiving change events from the database. Heartbeat messages might help decrease the number of change events that need to be re-sent when a connector restarts. To send heartbeat messages, set this property to a positive integer, which indicates the number of milliseconds between heartbeat messages. - **Inconsistent schema handling mode**: Specifies how the connector should react to binlog events that belong to a table missing from internal schema representation. Possible settings are: `fail`, `skip`, and `warn`. - `fail` (default): throws an exception that indicates the problematic event and its binlog offset, and causes the connector to stop. - `skip`: passes over the problematic event and does not log anything. - `warn`: logs the problematic event and its binlog offset and skips the event. - **Skip unparseable DDL**: A Boolean value that specifies whether the connector should ignore malformed or unknown database statements (`true`), or stop processing so a human can fix the issue (`false`). Defaults to `false`. Consider setting this to `true` to ignore unparseable statements. - **Store only captured tables DDL**: A Boolean value that specifies whether the connector records schema structures from all tables in a schema or database, or only from tables that are designated for capture. - `false` (default): During a database snapshot, the connector records the schema data for all non-system tables in the database, including tables that are not designated for capture. It’s best to retain the default setting. If you later decide to capture changes from tables that you did not originally designate for capture, the connector can easily begin to capture data from those tables, because their schema structure is already stored in the schema history topic. - `true`: During a database snapshot, the connector records the table schemas only for the tables from which Debezium captures change events. If you change the default value, and you later configure the connector to capture data from other tables in the database, the connector lacks the schema information that it requires to capture change events from the tables. - **Connection time zone**: Specifies how the server’s session time zone is determined. This property can take one of three values: LOCAL, SERVER, or a user-defined time zone. - **Snapshot select statement overrides data map**: A JSON object that maps fully-qualified table identifiers (`schemaName.tableName` or `databaseName.tableName`) to custom `SELECT` statements. The connector uses these statements during snapshots instead of the default `SELECT *` statement. Use this property for large append-only tables to resume a snapshot from a specific point if a previous attempt was interrupted. These values are sensitive and are masked in the configuration. #### NOTE Escape double quotes (`"`) in table or schema names. Use a single backslash (`\"`) in the UI and three backslashes (`\\\"`) in the CLI. **Schema Config** - **Key converter reference subject name strategy**: Set the subject reference name strategy for key. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy. **How should we handle data types?** - **Decimal handling mode**: Specifies how the connector should handle values for `DECIMAL` and `NUMERIC` columns. Possible settings are: `precise`, `double`, and `string`. - `precise` (default): represents values by using `java.math.BigDecimal` to represent values in binary form in change events. - `double`: represents values by using double values, which might result in a loss of precision but which is easier to use. - `string`: encodes values as formatted strings, which are easy to consume but semantic information about the real type is lost. - **Time precision mode**: Time, date, and timestamps can be represented with different kinds of precisions: - `adaptive_time_microseconds` (default): captures the date, datetime and timestamp values exactly as in the database using either millisecond, microsecond, or nanosecond precision values based on the database column’s type. An exception is `TIME` type fields, which are always captured as microseconds. - `connect`: always represents time and timestamp values by using Kafka Connect’s built-in representations for Time, Date, and Timestamp, which use millisecond precision regardless of the database columns’ precision. **Transforms** - **Single Message Transformations**: To add a new SMT, see [Add transforms](../single-message-transforms.md#cc-single-message-transforms-ui). For more information about unsupported SMTs, see [Unsupported transformations](../single-message-transforms.md#cc-single-message-transforms-unsupported-transforms). **Processing position** - **Set offsets**: Click **Set offsets** to define a specific offset for this connector to begin procession data from. For more information on managing offsets, see [Manage offsets](../offsets.md#connect-custom-offsets). For additional information about the Debezium SMTs **ExtractNewRecordState** and **EventRouter (Debezium)**, see [Debezium transformations](../single-message-transforms.md#cc-single-message-transforms-debezium-unwrap). For all property values and definitions, see [Configuration Properties](#cc-mysql-cdc-source-v2-config-properties). - Click **Continue**. ### Sizing Based on the number of topic partitions you select, you will be provided with a recommended number of tasks. 1. To change the number of tasks, use the Range Slider to select the desired number of tasks. 2. Click **Continue**. ### Review and Launch 1. Verify the connection details by previewing the running configuration. 2. After you’ve validated that the properties are configured to your satisfaction, click **Launch**. The status for the connector should go from **Provisioning** to **Running**. #### Step 5: Check the Kafka topic After the connector is running, verify that messages are populating your Kafka topic. #### NOTE A topic named `dbhistory..` is automatically created for `schema.history.internal.kafka.topic` with one partition. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section. ### Using the Confluent CLI Complete the following steps to set up and run the connector using the Confluent CLI. #### NOTE Make sure you have all your [prerequisites](#cc-mysql-cdc-source-v2-debezium-prereqs) completed. #### Step 1: List the available connectors Enter the following command to list available connectors: ```none confluent connect plugin list ``` #### Step 2: List the connector configuration properties Enter the following command to show the connector configuration properties: ```none confluent connect plugin describe ``` The command output shows the required and optional configuration properties. #### Step 3: Create the connector configuration file Create a JSON file that contains the connector configuration properties. The following examples show the required connector properties for both password and IAM role-based authentication. Using password authentication: ```none { "connector.class": "MySqlCdcSourceV2", "name": "MySqlCdcSourceV2Connector_0", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "****************", "kafka.api.secret": "****************************************************************", "database.hostname": "database-2..us-west-2.rds.amazonaws.com", "database.port": "3306", "database.user": "admin", "database.password": "**********", "topic.prefix": "mysql", "table.include.list":"employees.departments, "output.data.format": "JSON", "tasks.max": "1" } ``` Using IAM role-based authentication: ```none { "connector.class": "MySqlCdcSourceV2", "name": "MySqlCdcSourceV2Connector_0", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "****************", "kafka.api.secret": "****************************************************************", "database.hostname": "database-2..us-west-2.rds.amazonaws.com", "database.port": "3306", "database.user": "admin", "topic.prefix": "mysql", "table.include.list":"employees.departments, "output.data.format": "JSON", "tasks.max": "1", "authentication.method": "IAM Roles", "provider.integration.id": "dlz-f3a90de", "database.aws.region": "us-west-2" } ``` Note the following property definitions: * `"connector.class"`: Identifies the connector plugin name. * `"name"`: Sets a name for your new connector. * `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above). To use a [service account](../service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=`. To list the available service account resource IDs, use the following command: ```bash confluent iam service-account list ``` For example: ```bash confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2 ``` * `"database.hostname"`: IP address or hostname of the MySQL database server. * `"database.port"`: Port number of the MySQL database server. * `"database.user"`: The name of the MySQL database user that has the required authorization. * `"authentication.method"`: Set to `"Password"` to use password-based authentication or `"IAM Roles"` to use AWS IAM role-based authentication. * `"provider.integration.id"`: The ID of the provider integration resource that contains the IAM role configuration. * `"database.aws.region"`: The AWS region of the MySQL database server for RDS/Aurora. Only applicable when using IAM role-based authentication. * `"database.password"`: Password of the MySQL database user that has the required authorization. Only applicable when using password-based authentication. * `"topic.prefix"`: Provides a namespace for the particular database server/cluster that the connector is capturing changes from. * `"table.include.list"`: An optional, comma-separated list of fully-qualified table identifiers for the connector to monitor. By default, the connector monitors *all* non-system tables. A fully-qualified table name is in the form `databaseName.tableName`. This property cannot be used with the property `table.exclude.list`. * `"output.data.format"`: Sets the output Kafka record value format (data coming from the connector). Valid entries are **AVRO**, **JSON_SR**, **PROTOBUF**, or **JSON**. You must have Confluent Cloud Schema Registry configured if using a schema-based record format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). * `"tasks.max"`: Enter the number of [tasks](/platform/current/connect/concepts.html#tasks) in use by the connector. Organizations can run multiple connectors with a limit of one task per connector (that is, `"tasks.max": "1"`). #### NOTE To enable CSFLE or CSPE for data encryption, specify the following properties: * `csfle.enabled`: Flag to indicate whether the connector honors CSFLE or CSPE rules. * `sr.service.account.id`: A Service Account to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](../csfle.md#connect-csfle). **SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](../single-message-transforms.md#cc-single-message-transforms) documentation. For additional information about the Debezium SMTs **ExtractNewRecordState** and **EventRouter (Debezium)**, see [Debezium transformations](../single-message-transforms.md#cc-single-message-transforms-debezium-unwrap). See [Configuration Properties](#cc-mysql-cdc-source-v2-config-properties) for all properties and definitions. #### Step 4: Load the properties file and create the connector Enter the following command to load the configuration and start the connector: ```none confluent connect cluster create --config-file .json ``` For example: ```none confluent connect cluster create --config-file mysql-cdc-source-v2.json ``` Example output: ```none Created connector MySqlCdcSourceV2Connector_0 lcc-ix4dl ``` #### Step 5: Check the connector status Enter the following command to check the connector status: ```none confluent connect cluster list ``` Example output: ```none ID | Name | Status | Type +-----------+-------------------------------+---------+-------+ lcc-ix4dl | MySqlCdcSourceV2Connector_0 | RUNNING | source ``` #### Step 6: Check the Kafka topic. After the connector is running, verify that messages are populating your Kafka topic. #### NOTE A topic named `dbhistory..` is automatically created for `schema.history.internal.kafka.topic` with one partition. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section. ## Upgrading MySQL Database When you upgrade the MySQL database that the connector uses, you must take specific steps to protect against data loss and to ensure that the connector continues to operate. In general, the connector is resilient to interruptions caused by network failures and other outages. For example, when a database server that a connector monitors stops or crashes, after the connector re-establishes communication with the database server, it continues to read from the last position recorded in the offset. The connector retrieves information about the last recorded offset from the Connect offsets topic and queries the configured MySQL server to get events from the position recorded in the offset. #### NOTE This section includes content adapted from the [Debezium documentation](https://debezium.io/documentation/reference/stable/connectors/mysql.html), licensed under [Creative Commons Attribution 3.0](https://creativecommons.org/licenses/by/3.0/). Some details have been changed for the MySQL CDC Source V2 connector. There are various methods for upgrading a MySQL database. This section focuses on the in-place upgrade. For more details, see the [MySQL Upgrade documentation](https://dev.mysql.com/doc/refman/8.4/en/upgrade-binary-package.html). The Debezium MySQL CDC connector captures changes from the MySQL database by reading the binlog. During an in-place upgrade, the binlog files remain unaffected, allowing the connector to continue capturing changes from the upgraded server seamlessly. For guidance about how to perform an in-place MySQL database upgrade so that the connector can continue to capture events while minimizing the risk of data loss, see the following procedure: 1. Temporarily stop applications that write to the database, or put them into a read-only mode. 2. Back up the database. 3. Temporarily disable write access to the database. 4. Provide the connector with enough time to capture all event records that are written to the binlog. To verify that the connector has finished consuming entries from the binlog, check that the latest offset in the Kafka offsets topic is the same as the latest binlog position. Use the command - SHOW MASTER STATUS; to obtain the latest binlog position. To get the latest offset, refer to [Manage custom offsets](#cc-mysql-cdc-source-v2-debezium-custom-offsets). This step ensures that all change events that occurred before the downtime are accounted for, and that they are saved to Kafka. 5. Pause the connector. 6. Perform the in-place upgrade on the database. 7. Log in to the upgraded server and verify that the binlog which connector requires is present and that the server is setup for change data capture. 8. Resume the connector. 9. Restore write access to the database and restart any applications that write to the database. ## AWS IAM Authentication Setup To use AWS IAM role-based authentication with the MySQL CDC Source V2 connector, configure your MySQL database as described in the following steps: ### Common setup 1. Create a database user with AWS IAM authentication enabled: ```sql CREATE USER 'db-username'@'%' IDENTIFIED WITH AWSAuthenticationPlugin AS 'RDS'; ``` 2. Grant the necessary permissions to the user: ```sql GRANT REPLICATION CLIENT, REPLICATION SLAVE, SELECT ON *.* TO 'db-username'@'%'; ``` ### Amazon RDS setup For RDS instances, attach the following IAM permission policy to your IAM role that you have integrated with provider integration: ```json { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "rds-db:connect", "Resource": [ "arn:aws:rds-db:::dbuser:/" ] } ] } ``` Replace the `region`, `account-id`, `instance-identifier`, and `db-username` placeholders in the policy with the relevant values for your RDS instance. ### Aurora MySQL setup For Aurora MySQL clusters, complete the following steps: 1. Attach the following IAM permission policy to your IAM role that you have integrated with provider integration: ```json { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "rds-db:connect", "Resource": [ "arn:aws:rds-db:::dbuser:/" ] } ] } ``` Replace the placeholders in the policy with the relevant values for your Aurora cluster. 2. Add the following trust policy to your role: ```json { "Effect": "Allow", "Principal": { "Service": "rds.amazonaws.com" }, "Action": "sts:AssumeRole" } ``` 3. Configure the IAM role in your Aurora cluster: 1. In the AWS Console, navigate to your Aurora cluster. 2. Under **Connectivity & security**, locate the **Manage IAM roles** section. 3. Select **Add IAM roles to this cluster** and choose the IAM role you configured. 4. Add the IAM role to associate it with your Aurora cluster. #### NOTE * Ensure that the IAM role you use has the necessary permissions to access your RDS or Aurora instance. * For more information about using IAM authentication with RDS and Aurora, see [AWS documentation](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.IAMDBAuth.html). ## Moving from V1 to V2 Version 2 of this connector supports new features and has breaking changes that are not backward compatible with [version 1 of the connector](../cc-microsoft-sql-server-source-cdc-debezium.md#cc-microsoft-sql-server-source-cdc-debezium). To understand these changes and to plan for moving to version 2, see [Backward Incompatible Changes in Debezium CDC V2 Connectors](cc-debezium-v2-backward-incompatible-changes.md#cc-debezium-v2-backward-incompatible-changes-mysql). Given the backward-incompatible changes between version 1 and 2 of the CDC connectors, version 2 is being provided in a new set of CDC connectors on Confluent Cloud. You can provision either version 1 or version 2. However, note that eventually version 1 will be deprecated and no longer supported. Before exploring your options for moving from version 1 to 2, be sure to make the required changes documented in [Backward Incompatible Changes in Debezium CDC V2 Connectors](cc-debezium-v2-backward-incompatible-changes.md#cc-debezium-v2-backward-incompatible-changes-mysql). To get the offset in the following section, use the Confluent Cloud APIs. For more information, see [Connect offsets API reference](https://docs.confluent.io/cloud/current/ccloud/offsets-connect-v-1/), [Manage custom offsets](#cc-mysql-cdc-source-v2-debezium-custom-offsets), and [Manage Offsets for Fully Managed Connectors in Confluent Cloud](../offsets.md#connect-custom-offsets). **To move from version 1 to 2 (v1 to v2)** Use the following steps to migrate to version 2. Implement and validate any connector changes in a pre-production environment before promoting to production. 1. Pause the v1 connector. 2. [Get](https://docs.confluent.io/cloud/current/ccloud/get-connectv-1-connector-offsets/) the offset for the v1 connector. 3. Create the v2 connector using the offset from the previous step. ```none confluent connect cluster create [flags] ``` For example: Create a configuration file with connector configs and offsets. ```none { "name": "(connector-name)", "config": { ... // connector specific configuration }, "offsets": [ { "partition": { ... // connector specific configuration }, "offset": { ... // connector specific configuration } } ] } ``` Create a connector in the current or specified Kafka cluster context. ```none confluent connect cluster create --config-file config.json ``` For connectors that maintain a schema history topic, you must configure the schema history topic name in v2 to match the schema history topic name from the v1 connector. 4. [Delete](https://docs.confluent.io/cloud/current/ccloud/delete-connectv-1-connector/) the v1 connector. For more information, see [Manage Offsets for Fully Managed Connectors in Confluent Cloud](../offsets.md#connect-custom-offsets). ## Configuration Properties Use the following configuration properties with the fully managed connector. For self-managed connector property definitions and other details, see the connector docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html). ### How should we connect to your data? `name` : Sets a name for your connector.
* Type: string * Valid Values: A string at most 64 characters long * Importance: high ### Kafka Cluster credentials `kafka.auth.mode` : Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY * Importance: high `kafka.api.key` : Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high `kafka.service.account.id` : The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string * Importance: high `kafka.api.secret` : Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high ### Authentication method `authentication.method` : Select how you want to authenticate with the database. Valid options are `IAM Roles` and `Password`.
* Type: string * Default: Password * Valid Values: IAM Roles, Password * Importance: high `provider.integration.id` : Select an existing integration that has access to your resource.
* Type: string * Importance: high `secret.manager.enabled` : Fetch sensitive configuration values from a secret manager.
* Type: boolean * Default: false * Importance: high ### Secret manager configuration `secret.manager` : Select the secret manager to use for retrieving sensitive data.
* Type: string * Importance: high `secret.manager.managed.configs` : Select the configurations to fetch their values from the secret manager.
* Type: list * Importance: high `secret.manager.provider.integration.id` : Select an existing provider integration that has access to your secret manager.
* Type: string * Importance: high ### How should we connect to your database? `database.aws.region` : The AWS region of the MySQL database server for RDS/Aurora.
* Type: string * Importance: high `database.hostname` : IP address or hostname of the MySQL database server.
* Type: string * Valid Values: Must match the regex `^[A-Za-z0-9_.-]+$` * Importance: high `database.port` : Port number of the MySQL database server.
* Type: int * Valid Values: [0,…,65535] * Importance: high `database.user` : The name of the MySQL database user that has the required authorization.
* Type: string * Importance: high `database.password` : Password for the MySQL database user that has the required authorization
* Type: password * Importance: high `database.ssl.mode` : Whether to use an encrypted connection to the MySQL server. Possible settings are: disabled, preferred, required, verify_ca, and verify_identity.
disabled specifies the use of an unencrypted connection.
preferred establishes an encrypted connection if the server supports secure connections. If the server does not support secure connections, falls back to an unencrypted connection.
required establishes an encrypted connection or fails if one cannot be made for any reason.
verify_ca The connector behaves as when you set the required option, but it also verifies the server TLS certificate against the configured Certificate Authority (CA) certificates. If the server TLS certificate does not match any valid CA certificates, the connector fails.
verify_identity The connector behaves as when you set the verify_ca option, but it also verifies that the server certificate matches the host of the remote connection.
* Type: string * Default: preferred * Importance: high `database.ssl.keystore` : Path to the SSL keystore file for MySQL connection. Only needed when SSL certificate verification is required (verify_ca or verify_identity modes).
* Type: password * Importance: medium `database.ssl.keystore.password` : Password for the SSL keystore file for MySQL connection. Only needed when SSL certificate verification is required and the keystore is password-protected.
* Type: password * Importance: medium `database.ssl.truststore` : Path to the SSL truststore file for MySQL connection. Only needed when SSL certificate verification is required (verify_ca or verify_identity modes).
* Type: password * Importance: medium `database.ssl.truststore.password` : Password for the SSL truststore file for MySQL connection. Only needed when SSL certificate verification is required and the truststore is password-protected.
* Type: password * Importance: medium ### Output messages `output.data.format` : Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, PROTOBUF, or JSON. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF
* Type: string * Default: JSON * Importance: high `output.key.format` : Sets the output Kafka record key format. Valid entries are AVRO, JSON_SR, PROTOBUF, STRING or JSON. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF
* Type: string * Default: JSON * Valid Values: AVRO, JSON, JSON_SR, PROTOBUF, STRING * Importance: high `after.state.only` : Controls whether the generated Kafka record should contain only the state of the row after the event occurred.
* Type: boolean * Default: false * Importance: low `tombstones.on.delete` : Controls whether a delete event is followed by a tombstone event. The following values are possible:
`true`: For each delete operation, the connector emits a delete event and a subsequent tombstone event.
`false`: For each delete operation, the connector emits only a delete event.
After a source record is deleted, a tombstone event (the default behavior) enables Kafka to completely delete all events that share the key of the deleted row in topics that have log compaction enabled.
* Type: boolean * Default: true * Importance: medium `after.state.only.replace.null.with.default` : Applies only when After-state only is enabled. When set to true, null field values are replaced by the source-defined default values when the connector rewrites the record. Set to false to preserve null field values (for example, so that clearing a column value or removing a column default is reflected as null in the output record).
* Type: boolean * Default: true * Importance: low ### How should we name your topic(s)? `topic.prefix` : Topic prefix that provides a namespace (logical server name) for the particular MySQL database server or cluster in which Debezium is capturing changes. The prefix should be unique across all other connectors, since it is used as a topic name prefix for all Kafka topics that receive records from this connector. Only alphanumeric characters, hyphens, dots and underscores must be used. The connector automatically creates Kafka topics using the naming convention: ...
* Type: string * Importance: high `schema.history.internal.kafka.topic` : The name of the topic for the database schema history. A new topic with provided name is created, if it doesn’t already exist. If the topic already exists, ensure that it has a single partition, infinite retention period and is not in use by any other connector. If no value is provided, the name defaults to `dbhistory..`.
* Type: string * Default: dbhistory.${topic.prefix}.{{.logicalClusterId}} * Importance: high ### How should we configure the topic(s)? `topic.creation.topic_prefix_match.partitions` : Number of partitions for Kafka topics auto-created by the connector for topics whose name starts with `topic.prefix.`. Kafka preserves message ordering only within a partition. The connector keys each change event by the changed record’s primary key, and Kafka routes all events with the same key to the same partition - so events sharing a key stay in order, while events with different keys may be spread across partitions and lose their relative order. Tables that have no primary key produce unkeyed events, which Kafka distributes across all partitions, so any value above 1 removes ordering guarantees for those tables entirely. Keep this at 1 (the default) if you need all change events in a topic delivered in strict order.
* Type: int * Default: 1 * Valid Values: [1,…] * Importance: high `topic.creation.topic_prefix_match.cleanup.policy` : Cleanup policy applied to Kafka topics auto-created by the connector for topics whose name starts with `topic.prefix.`. `compact` retains only the latest value per key (typical for change-data-capture topics); `delete` ages records out based on retention settings.
* Type: string * Default: delete * Valid Values: compact, compact,delete, delete * Importance: high ### Storage `topic.creation.topic_prefix_match.retention.ms` : Time-based retention, in milliseconds, applied to Kafka topics auto-created by the connector for topics whose name starts with `topic.prefix.`. Use `-1` for infinite retention.
* Type: long * Default: 604800000 (7 days) * Valid Values: [-1,…] * Importance: high `topic.creation.topic_prefix_match.retention.bytes` : Size-based retention, in bytes, applied to Kafka topics auto-created by the connector for topics whose name starts with `topic.prefix.`. Use `-1` for unlimited size.
* Type: long * Default: -1 * Valid Values: [-1,…] * Importance: high ### Database config `signal.data.collection` : Fully-qualified name of the data collection that needs to be used to send signals to the connector. Use the following format to specify the fully-qualified collection name: databaseName.tableName
* Type: string * Importance: medium ### Connector config `snapshot.mode` : Specifies the criteria for running a snapshot when the connector starts. Possible settings are: initial, initial_only, when_needed, never, schema_only\`(deprecated), \`no_data, schema_only_recovery\`(deprecated), \`recovery.
initial - the connector runs a snapshot only when no offsets have been recorded for the logical server name.
initial_only - the connector runs a snapshot only when no offsets have been recorded for the logical server name and then stops; i.e. it will not read change events from the binlog.
when_needed - the connector runs a snapshot upon startup whenever it deems it necessary. That is, when no offsets are available, or when a previously recorded offset specifies a binlog location or GTID that is not available in the server.
never - the connector never uses snapshots. Upon first startup with a logical server name, the connector reads from the beginning of the binlog. Configure this behavior with care. It is valid only when the binlog is guaranteed to contain the entire history of the database.
schema_only - Deprecated, use no_data instead.
no_data - the connector runs a snapshot of the schemas and not the data. This setting is useful when you do not need the topics to contain a consistent snapshot of the data but need them to have only the changes since the connector was started.
schema_only_recovery - Deprecated, use recovery instead.
recovery - recovery setting for a connector that has already been capturing changes. When you restart the connector, this setting enables recovery of a corrupted or lost database schema history topic. You might set it periodically to “clean up” a database schema history topic that has been growing unexpectedly. Database schema history topics require infinite retention.
* Type: string * Default: initial * Valid Values: initial, initial_only, never, no_data, recovery, schema_only, schema_only_recovery, when_needed * Importance: medium `snapshot.locking.mode` : Controls whether and how long the connector holds the global MySQL read lock, which prevents any updates to the database, while the connector is performing a snapshot. Possible settings are: minimal, minimal_percona, extended, and none.
minimal - the connector holds the global read lock for just the initial portion of the snapshot, while the database schemas and other metadata are being read. The remaining work in a snapshot involves selecting all rows from each table. This is accomplished using a REPEATABLE READ transaction, even when the lock is no longer held and other MySQL clients are updating the database.
minimal_percona - similar to minimal mode except the connector uses a (Percona-specific) backup lock. This mode does not flush tables to disk, is not blocked by long-running reads, and is available only in Percona Server.
extended - blocks all writes for the duration of the snapshot. Use this setting if there are clients that are submitting operations that MySQL excludes from REPEATABLE READ semantics.
none - prevents the connector from acquiring any table locks during the snapshot. While this setting is allowed with all snapshot modes, it is safe to use if and only if no schema changes are happening while the snapshot is running. For tables defined with MyISAM engine, the tables would still be locked despite this property being set as MyISAM acquires a table lock. This behavior is unlike InnoDB engine, which acquires row level locks.
* Type: string * Default: minimal * Valid Values: extended, minimal, minimal_percona, none * Importance: low `database.include.list` : An optional, comma-separated list of regular expressions that match the names of the databases for which to capture changes. The connector does not capture changes in any database whose name is not in this list. By default, the connector captures changes in all databases.
To match the name of a database, Debezium applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire name string of the database; it does not match substrings that might be present in a database name.
* Type: list * Importance: medium `database.exclude.list` : A comma-separated list of regular expressions that match the names of databases from which you do not want the connector to capture changes. The connector captures changes in any database that is not named in the `database.exclude.list`
* Type: list * Importance: medium `table.include.list` : An optional, comma-separated list of regular expressions that match fully-qualified table identifiers for tables whose changes you want to capture. When this property is set, the connector captures changes only from the specified tables. Each identifier is of the form database.tableName. By default, the connector captures changes in every non-system table in each schema whose changes are being captured.
To match the name of a table, Debezium applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire identifier for the table; it does not match substrings that might be present in a table name.
If you include this property in the configuration, do not also set the `table.exclude.list` property.
* Type: list * Importance: medium `table.exclude.list` : An optional, comma-separated list of regular expressions that match fully-qualified table identifiers for tables whose changes you do not want to capture. Each identifier is of the form database.tableName. When this property is set, the connector captures changes from every table that you do not specify.
To match the name of a table, Debezium applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire identifier for the table; it does not match substrings that might be present in a table name.
If you include this property in the configuration, do not set the `table.include.list` property.
* Type: list * Importance: medium `event.processing.failure.handling.mode` : Specifies how the connector should react to exceptions during processing of events. Possible settings are `fail`, `skip`, and `warn`.
`fail`: Propagates the exception, indicates the offset of the problematic event, and causes the connector to stop.
`warn`: Logs the offset of the problematic event, skips that event, and continues processing.
`skip`: Skips the problematic event and continues processing.
* Type: string * Default: fail * Valid Values: fail, skip, warn * Importance: low `column.exclude.list` : An optional, comma-separated list of regular expressions that match the fully-qualified names of columns to exclude from change event record values. Fully-qualified names for columns are of the form `databaseName.tableName.columnName`.
To match the name of a column, Debezium applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire name string of the column; it does not match substrings that might be present in a column name.
* Type: list * Importance: medium `schema.name.adjustment.mode` : Specifies how schema names should be adjusted for compatibility with the message converter used by the connector. The following values are possible:
`none`: Does not apply any adjustment.
`avro`: Replaces the characters that cannot be used in the Avro type name with underscore.
`avro_unicode`: Replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode such as, `_uxxxx`. Note that `_` is an escape sequence like backslash in Java.
* Type: string * Default: none * Valid Values: avro, avro_unicode, none * Importance: medium `field.name.adjustment.mode` : Specifies how field names should be adjusted for compatibility with the message converter used by the connector. The following values are possible:
`none`: Does not apply any adjustment.
`avro`: Replaces the characters that cannot be used in the Avro type name with underscore.
`avro_unicode`: Replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode such as, `_uxxxx`. Note that `_` is an escape sequence like backslash in Java.
* Type: string * Default: none * Valid Values: avro, avro_unicode, none * Importance: medium `heartbeat.interval.ms` : Controls how frequently the connector sends heartbeat messages to a Kafka topic. The behavior of default value 0 is that the connector does not send heartbeat messages. Heartbeat messages are useful for monitoring whether the connector is receiving change events from the database. Heartbeat messages might help decrease the number of change events that need to be re-sent when a connector restarts. To send heartbeat messages, set this property to a positive integer, which indicates the number of milliseconds between heartbeat messages.
* Type: int * Default: 0 * Valid Values: [0,…] * Importance: low `snapshot.select.statement.overrides.data.map` : A JSON object that maps fully-qualified table identifiers (`schemaName.tableName` or `databaseName.tableName`) to custom SELECT statements. The connector uses these statements during snapshots instead of the default SELECT \* query. Use this property for large append-only tables to resume a snapshot from a specific point if a previous attempt was interrupted. These values are sensitive and are masked in the configuration.
Note that you must escape double quotes (`"`) in table or schema names. Use a single backslash (`\"`) in the UI and three backslashes (`\\\"`) in the CLI.
* Type: password * Importance: medium `inconsistent.schema.handling.mode` : Specifies how the connector should react to binlog events that belong to a table missing from internal schema representation. Possible settings are: fail, skip, and warn.
fail - throws an exception that indicates the problematic event and its binlog offset, and causes the connector to stop.
skip - passes over the problematic event and does not log anything.
warn - logs the problematic event and its binlog offset and skips the event.
* Type: string * Default: fail * Valid Values: fail, skip, warn * Importance: medium `schema.history.internal.skip.unparseable.ddl` : A Boolean value that specifies whether the connector should ignore malformed or unknown database statements (true), or stop processing so a human can fix the issue (false). Defaults to false. Consider setting this to true to ignore unparseable statements.
* Type: boolean * Default: false * Importance: low `schema.history.internal.store.only.captured.tables.ddl` : A Boolean value that specifies whether the connector records schema structures from all tables in a schema or database, or only from tables that are designated for capture. Defaults to false.
false - During a database snapshot, the connector records the schema data for all non-system tables in the database, including tables that are not designated for capture. It’s best to retain the default setting. If you later decide to capture changes from tables that you did not originally designate for capture, the connector can easily begin to capture data from those tables, because their schema structure is already stored in the schema history topic.
true - During a database snapshot, the connector records the table schemas only for the tables from which Debezium captures change events. If you change the default value, and you later configure the connector to capture data from other tables in the database, the connector lacks the schema information that it requires to capture change events from the tables.
* Type: boolean * Default: false * Importance: low `driver.connectionTimeZone` : Specifies how the server’s session time zone is determined. This property can take one of three values: LOCAL, SERVER, or a user-defined time zone.
* Type: string * Importance: low ### Schema Config `schema.context.name` : Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
* Type: string * Default: default * Importance: medium `key.converter.reference.subject.name.strategy` : Set the subject reference name strategy for key. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: high ### How should we handle data types? `decimal.handling.mode` : Specifies how the connector should handle DECIMAL and NUMERIC columns. You can set one of the following options:
`precise`: Represents values by using `java.math.BigDecimal` to represent values in binary form in change events.
`double`: Represents values by using double values, which might result in a loss of precision but which is easier to use.
`string`: encodes values as formatted strings, which are easy to consume but semantic information about the real type is lost.
* Type: string * Default: precise * Valid Values: double, precise, string * Importance: medium `time.precision.mode` : Time, date, and timestamps can be represented with different kinds of precisions:
adaptive_time_microseconds captures the date, datetime and timestamp values exactly as in the database using either millisecond, microsecond, or nanosecond precision values based on the database column’s type. An exception is TIME type fields, which are always captured as microseconds.
connect always represents time and timestamp values by using Kafka Connect’s built-in representations for Time, Date, and Timestamp, which use millisecond precision regardless of the database columns’ precision.
* Type: string * Default: adaptive_time_microseconds * Valid Values: adaptive_time_microseconds, connect * Importance: medium ### Number of tasks for this connector `tasks.max` : Maximum number of tasks for the connector.
* Type: int * Valid Values: [1,…,1] * Importance: high ### Additional Configs `column.include.list` : A comma-separated list of regular expressions that match the fully-qualified names of columns that should be included in change event record values. Fully-qualified names for columns are of the form databaseName.tableName.columnName. Do not set `column.exclude.list` if you set this property.
* Type: list * Importance: low `column.propagate.source.type` : A comma-separated list of regular expressions matching fully-qualified names of columns that adds the column’s original type and original length as parameters to the corresponding field schemas in the emitted change records. When this property is set, the connector adds the following fields to the schema of event records with prefix `__debezium.source.column`. These parameters propagate a column’s original type name and length (for variable-width types), respectively. Include ‘.\*’ to match all column types.’
* Type: list * Importance: low `gtid.source.excludes` : A comma-separated list of regular expressions that match source domain IDs in the GTID set that the connector uses to find the binlog position on the MySQL server. When this property is set, the connector uses only the GTID ranges that have source UUIDs that do not match any of the specified exclude patterns.
* Type: string * Importance: low `gtid.source.includes` : A comma-separated list of regular expressions that match source domain IDs in the GTID set used that the connector uses to find the binlog position on the MySQL server. When this property is set, the connector uses only the GTID ranges that have source UUIDs that match one of the specified include patterns.
* Type: string * Importance: low `header.converter` : The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string * Importance: low `message.key.columns` : A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ‘:’, where the fully qualified table name could be defined as `.` and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table’s primary key column(s) will be used as message key. Example: `inventory.customers:pk1,pk2;(.*).purchaseorders:pk3,pk4`
* Type: string * Importance: low `notification.enabled.channels` : List of notification channels names that are enabled. The following channels are available: `log` and `sink`. When `sink` is enabled, the connector sends notifications to a topic specified by the `notification.sink.topic.name` property.
* Type: list * Importance: low `notification.sink.topic.name` : The name of the topic for the notifications. This is required in case `sink` is in the list of enabled channels. If you set this name so that it starts with your topic prefix followed by a period, the topic uses the same partition, retention, and cleanup settings as your other topics with that prefix; otherwise it is created with a single partition.
* Type: string * Importance: low `producer.override.compression.type` : The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
* Type: string * Importance: low `producer.override.linger.ms` : The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: [https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms).
* Type: long * Valid Values: [100,…,1000] * Importance: low `signal.enabled.channels` : A comma-separated list of channel names that are enabled for the connector. If not set, the connector enables only the `source` channel by default. Supported values are:
`source` (default): Signals are read from a signaling table in the source database.
`kafka`: Signals are consumed from a Kafka topic.
* Type: list * Importance: low `signal.kafka.topic` : The name of the Kafka topic that the connector monitors for ad hoc signals. Note that you can currently send signal messages to this topic via the Confluent CLI. Note that signal.kafka.topic must have exactly 1 partition as the connector’s signal consumer reads only from partition-0. Any signal messages routed to other partitions are silently ignored in a multi-partition topic.
* Type: string * Importance: low `snapshot.include.collection.list` : A comma-separated list of regular expressions that match the fully-qualified names (.) of the tables to include in a snapshot. If not explicitly set, the connector defaults to snapshotting all tables listed in table.include.list. The specified items must be named in the connector’s table.include.list property. This property takes effect only if the connector’s snapshot.mode property is set to a value other than never.
* Type: list * Importance: low `value.converter.allow.optional.map.keys` : Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.auto.register.schemas` : Specify if the Serializer should attempt to register the Schema.
* Type: boolean * Importance: low `value.converter.connect.meta.data` : Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.avro.schema.support` : Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.protobuf.schema.support` : Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.flatten.unions` : Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.index.for.unions` : Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.struct.for.nulls` : Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.int.for.enums` : Whether to represent enums as integers. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.latest.compatibility.strict` : Verify latest subject version is backward compatible when use.latest.version is true.
* Type: boolean * Importance: low `value.converter.object.additional.properties` : Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.optional.for.nullables` : Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.optional.for.proto2` : Whether proto2 optionals are supported. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.scrub.invalid.names` : Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.
* Type: boolean * Importance: low `value.converter.use.latest.version` : Use latest version of schema in subject for serialization when auto.register.schemas is false.
* Type: boolean * Importance: low `value.converter.use.optional.for.nonrequired` : Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.nullables` : Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.raw.primitives` : Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `incremental.snapshot.chunk.size` : The maximum number of rows that the connector fetches and reads into memory during an incremental snapshot chunk. Increasing the chunk size improves efficiency by running fewer, larger snapshot queries. However, larger chunk sizes also require more memory to buffer the snapshot data. Adjust the chunk size to a value that provides the best performance in your environment.
* Type: int * Default: 1024 * Valid Values: [1,…,1024] * Importance: medium `read.only` : Controls whether the connector writes watermarks to the signal data collection to track incremental snapshot progress. Set the value to `true` to enable the connector to use an incremental snapshot watermarking strategy that does not require writing to the signal data collection (useful for read-only database connections).
* Type: boolean * Default: false * Importance: medium `bigint.unsigned.handling.mode` : Specify how BIGINT UNSIGNED columns should be represented in change events, including: ‘precise’ uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect’s ‘org.apache.kafka.connect.data.Decimal’ type; ‘long’ (the default) represents values using Java’s ‘long’, which may not offer the precision but will be far easier to use in consumers.
* Type: string * Default: long * Importance: low `binary.handling.mode` : Specify how binary (blob, binary, etc.) columns should be represented in change events, including: ‘bytes’ represents binary data as byte array (default); ‘base64’ represents binary data as base64-encoded string; ‘base64-url-safe’ represents binary data as base64-url-safe-encoded string; ‘hex’ represents binary data as hex-encoded (base16) string
* Type: string * Default: bytes * Importance: low `connect.timeout.ms` : Maximum time to wait after trying to connect to the database before timing out, given in milliseconds. Defaults to 30 seconds (30,000 ms).
* Type: long * Default: 30000 (30 seconds) * Importance: low `enable.time.adjuster` : MySQL allows user to insert year value as either 2-digit or 4-digit. In case of two digit the value is automatically mapped into 1970 - 2069. The values `false` - delegates the implicit conversion to the database. `true` - (the default) Debezium makes the conversion
* Type: boolean * Default: true * Importance: low `errors.tolerance` : Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
* Type: string * Default: none * Importance: low `event.converting.failure.handling.mode` : Specify how failures during converting of event should be handled, including: ‘fail’ throw an exception that the column of event conversion is failed with unmatched schema type, causing the connector to be stopped. it could need schema recovery to covert successfully; ‘warn’ (the default) the value of column of event that conversion failed will be null and be logged with warn level; ‘skip’ the value of column of event that conversion failed will be null and be logged with debug level.
* Type: string * Default: warn * Importance: low `include.query` : Determines whether the connector should include the original SQL query that generated the change event. Note: This option requires MySQL be configured with the `binlog_rows_query_log_events` option set to `ON`. Query will not be present for events generated from snapshot. WARNING: Enabling this option may expose tables or fields explicitly excluded or masked by including the original SQL statement in the change event. For this reason the default value is `false`.
* Type: boolean * Default: false * Importance: low `include.schema.changes` : Whether the connector should publish changes in the database schema to a Kafka topic with the same name as the database server ID. Each schema change will be recorded using a key that contains the database name and whose value include logical description of the new schema and optionally the DDL statement(s). The default is `false`. This is independent of how the connector internally records database schema history.
* Type: boolean * Default: false * Importance: low `incremental.snapshot.allow.schema.changes` : Detect schema change during an incremental snapshot and re-select a current chunk to avoid locking DDLs. Note that changes to a primary key are not supported and can cause incorrect results if performed during an incremental snapshot. Another limitation is that if a schema change affects only columns’ default values, then the change won’t be detected until the DDL is processed from the binlog stream. This doesn’t affect the snapshot events’ values, but the schema of snapshot events may have outdated defaults.
* Type: boolean * Default: false * Importance: low `incremental.snapshot.watermarking.strategy` : Specify the strategy used for watermarking during an incremental snapshot: ‘INSERT_INSERT’ both open and close signal is written into signal data collection (default); ‘INSERT_DELETE’ only open signal is written on signal data collection, the close will delete the relative open signal.
* Type: string * Default: INSERT_INSERT * Importance: low `key.converter.key.schema.id.serializer` : The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `key.converter.key.subject.name.strategy` : How to construct the subject name for key schema registration.
* Type: string * Default: TopicNameStrategy * Importance: low `max.batch.size` : Maximum size of each batch of events that the connector processes. Defaults to 2048 with the allowed range is from 1 to 5000.
* Type: int * Default: 2048 * Valid Values: [1,…,5000] * Importance: low `poll.interval.ms` : Time to wait for new change events to appear after receiving no events, given in milliseconds. Defaults to 500 ms.
* Type: long * Default: 500 * Valid Values: [200,…] * Importance: low `provide.transaction.metadata` : Determines whether the connector generates events with transaction boundaries and enriches change event envelopes with transaction metadata. When enabled, the connector creates a dedicated transaction metadata topic. Its name starts with your topic prefix, so it uses the same partition, retention, and cleanup settings as your other topics with that prefix.
* Type: boolean * Default: false * Importance: low `schema.history.internal.kafka.create.timeout.ms` : The number of milliseconds to wait while create kafka history topic using Kafka admin client.
* Type: long * Default: 30000 (30 seconds) * Importance: low `schema.history.internal.kafka.query.timeout.ms` : The number of milliseconds to wait while fetching cluster information using Kafka admin client.
* Type: long * Default: 3000 (3 seconds) * Importance: low `schema.history.internal.kafka.recovery.poll.interval.ms` : An integer value that specifies the maximum number of milliseconds the connector should wait during startup/recovery while polling for persisted data. The default is 100ms.
* Type: long * Default: 100 * Valid Values: [100,…] * Importance: low `schema.history.internal.store.only.captured.databases.ddl` : Controls what DDL will Debezium store in database schema history. By default (false) Debezium will store all incoming DDL statements. If set to true, then only DDL that manipulates a table from captured schema/database will be stored.
* Type: boolean * Default: false * Importance: low `skip.messages.without.change` : Enable to skip publishing messages when there is no change in included columns. This would essentially filter messages to be sent when there is no change in columns included as per `column.include.list` or `column.exclude.list`. Set the value to `true` to prevent the connector from capturing records when no changes are present in the included columns.
* Type: boolean * Default: false * Importance: low `skipped.operations` : The comma-separated list of operations to skip during streaming, defined as: ‘c’ for inserts/create; ‘u’ for updates; ‘d’ for deletes, ‘t’ for truncates, and ‘none’ to indicate nothing skipped. By default, only truncate operations will be skipped.
* Type: list * Default: t * Importance: low `snapshot.delay.ms` : An interval in milliseconds that the connector should wait before performing a snapshot when the connector starts. Defaults to 0 ms.
* Type: long * Default: 0 * Valid Values: [0,…] * Importance: low `snapshot.lock.timeout.ms` : The maximum number of millis to wait for table locks at the beginning of a snapshot. If locks cannot be acquired in this time frame, the snapshot will be aborted. Defaults to 10 seconds.
* Type: long * Default: 10000 (10 seconds) * Importance: low `snapshot.tables.order.by.row.count` : Controls the order in which tables are processed in the initial snapshot. A ‘descending’ value will order the tables by row count descending. A ‘ascending’ value will order the tables by row count ascending. A value of ‘disabled’ (the default) will disable ordering by row count.
* Type: string * Default: disabled * Importance: low `streaming.delay.ms` : A delay period after the snapshot is completed and the streaming begins, given in milliseconds. This delay helps prevent re-snapshotting in case the connector fails during the transition to streaming. Defaults to 60000 ms.
* Type: long * Default: 60000 (1 minute) * Valid Values: [0,…] * Importance: low `table.ignore.builtin` : A Boolean value that specifies whether built-in system tables should be ignored. This applies regardless of the table include and exclude lists.
* Type: boolean * Default: true * Importance: low `topic.heartbeat.prefix` : Specifies the prefix of the heartbeat topic to which the connector sends heartbeat messages. The topic name has this pattern: `.`. Defaults to `__debezium-heartbeat-{{.logicalClusterId}}`. By default the heartbeat topic does not start with your topic prefix, so it is created with a single partition. If you change this prefix so that the heartbeat topic name starts with your topic prefix followed by a period, it instead uses the same partition, retention, and cleanup settings as your other topics with that prefix. Keep the default unless you have a specific reason to change it.
* Type: string * Default: \_\_debezium-heartbeat-{{.logicalClusterId}} * Importance: low `topic.transaction` : Controls the name of the topic to which the connector sends transaction metadata messages. The final transaction topic name has this pattern: `.`. Defaults to `{{.logicalClusterId}}.transaction`. Because this topic’s name always starts with your topic prefix, it uses the same partition, retention, and cleanup settings as your other topics with that prefix.
* Type: string * Default: {{.logicalClusterId}}.transaction * Importance: low `use.nongraceful.disconnect` : Whether to use socket.setSoLinger(true, 0) when BinaryLogClient keepalive thread triggers a disconnect for a stale connection.
* Type: boolean * Default: false * Importance: low `value.converter.decimal.format` : Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string * Default: BASE64 * Importance: low `value.converter.flatten.singleton.unions` : Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.ignore.default.for.nullables` : When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.reference.subject.name.strategy` : Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: low `value.converter.replace.null.with.default` : Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.
* Type: boolean * Default: true * Importance: low `value.converter.schemas.enable` : Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.
* Type: boolean * Default: false * Importance: low `value.converter.value.schema.id.serializer` : The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `value.converter.value.subject.name.strategy` : Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string * Default: TopicNameStrategy * Importance: low ### Auto-restart policy `auto.restart.on.user.error` : Enable connector to automatically restart on user-actionable errors.
* Type: boolean * Default: true * Importance: medium ## Monitoring and Troubleshooting ### Connector appears as RUNNING but is not processing data After a **task restart**, the MySQL CDC Source v2 connector performs **schema history recovery** before it resumes change data capture (CDC). During this phase: - The connector status in the Confluent Cloud Console appears as `RUNNING`. - However, **no records will be emitted** until schema history recovery is complete. - This is expected behavior and does **not** indicate an error. To monitor the progress of schema history recovery, use the following Confluent Cloud metric: ```none mysql_cdc_source_connector_schema_history_status ``` This is a gauge metric with the following values: - `0` — Schema recovery is **stopped** - `1` — Schema recovery is **in progress** - `2` — Schema recovery is **complete** Once the metric reaches `2`, the connector resumes streaming change events to the Kafka topic. Schema history recovery occurs on every **task restart**, including those triggered by configuration changes, scaling operations, or infrastructure events. While the connector shows as `RUNNING`, schema history recovery happens as a separate process after the task initialization. For more information on available metrics, refer to the [Confluent Cloud Metrics API reference](https://api.telemetry.confluent.cloud/docs/descriptors/datasets/cloud). ## FAQs Find answers to frequently asked questions about the MySQL CDC Source V2 (Debezium) connector for Confluent Cloud. ### How do I connect to a self-hosted MySQL database using private networking? To connect the fully managed MySQL CDC Source V2 connector to a self-hosted MySQL database over private networking, configure egress endpoints: 1. **Create a PrivateLink egress endpoint**: In the Cloud Console, navigate to your environment and create an egress PrivateLink endpoint that connects to your MySQL database. 2. **Configure the endpoint service**: Ensure your MySQL database is accessible through an Amazon PrivateLink endpoint service, Azure Private Link service, or Google Cloud Private Service Connect. 3. **Update the connector configuration**: In the `database.hostname` property, use the DNS name of the egress endpoint instead of a public IP address or hostname. 4. **Verify connectivity**: Ensure the egress endpoint status is `Ready` before creating the connector. You can verify this in the Cloud Console under **Networking > Egress**. #### IMPORTANT The egress endpoint must be in the same region as your Kafka cluster. Cross-region egress is not supported. For detailed setup instructions, see [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud). ### What MySQL binlog configuration is required for the connector? The MySQL CDC Source V2 connector requires specific binlog configuration on your MySQL server to capture change events. Required binlog settings: 1. Enable binary logging by adding the following to the MySQL server configuration file: ```ini [mysqld] server-id=223344 # must be unique across your MySQL servers log_bin=mysql-bin binlog_format=ROW binlog_row_image=FULL binlog_expire_logs_seconds=604800 # 7 days, recommended ``` 2. **Verify the binlog configuration**: ```sql SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'binlog_row_image'; ``` 3. Enable GTID mode, which is recommended for more reliable failover and recovery: ```ini gtid_mode=ON enforce_gtid_consistency=ON ``` If you run MySQL on Amazon RDS, you must enable automated backups for the database instance for binary logging to occur. Without automated backups, the binlog is disabled even if you apply these settings. For more about enabling the MySQL binlog, see [Enabling the binlog](https://debezium.io/documentation/reference/3.2/connectors/mysql.html#enable-mysql-binlog) in the Debezium documentation. ### Why is my connector failing with `Access denied` errors? Authentication errors typically indicate missing MySQL user privileges or incorrect credentials. Common error messages: ```text Access denied for user 'connector_user'@'%' Access denied; you need (at least one of) the REPLICATION SLAVE privilege(s) ``` Required MySQL privileges: 1. **Grant the required privileges**: Create the MySQL user and grant the privileges required for change data capture. For the required privileges and the user-creation steps, see [Creating a MySQL user](https://debezium.io/documentation/reference/3.2/connectors/mysql.html#mysql-creating-user) in the Debezium documentation. 2. **Verify privileges**: ```sql SHOW GRANTS FOR 'connector_user'@'%'; ``` 3. **Verify password authentication**: Ensure the `database.user` and `database.password` properties match the MySQL user credentials exactly. 4. **Check host restrictions**: Verify the user is allowed to connect from the connector’s IP address. Use `'%'` for any host or specify the connector’s IP range. ### How do I configure AWS IAM authentication for RDS MySQL? The MySQL CDC Source V2 connector supports Amazon IAM role-based authentication for RDS and Aurora MySQL using Confluent Provider Integration. To use it, set the following connector properties: * Set **Authentication method** to `AWS IAM Roles`. * Select the **Provider Integration** that has access to your database. * Set **Database AWS region** to the region of your RDS or Aurora instance. Do not set the database password when using IAM authentication. For the database user, IAM policy, and Provider Integration setup steps, see [AWS IAM Authentication Setup](#cc-mysql-cdc-source-v2-debezium-aws-iam-auth). ### Why is my initial snapshot taking too long or failing? Large database snapshots can be slow or fail due to table size, network latency, or resource constraints. Solutions: 1. **Use an incremental snapshot**: An incremental snapshot captures table data in chunks and does not hold the global read lock that a blocking initial snapshot uses. Configure `signal.data.collection` with a signaling table, and then send an incremental snapshot signal. For details, see [Debezium incremental snapshots](https://debezium.io/documentation/reference/3.2/connectors/mysql.html#debezium-mysql-incremental-snapshots). 2. **Limit the snapshot scope**: Use table filtering to snapshot only the tables you need: ```json { "table.include.list": "mydb.orders,mydb.customers" } ``` 3. **Monitor snapshot progress**: Check the connector logs in the Cloud Console for snapshot status: * Look for log messages indicating which tables are being snapshotted. * Check for `Snapshot step` progress messages. #### NOTE The connector establishes a consistent snapshot using a global read lock (`FLUSH TABLES WITH READ LOCK`). How long the lock is held depends on the `snapshot.locking.mode` property. The `minimal` mode holds the lock only while reading schema and metadata, whereas `extended` blocks writes for the entire snapshot. For details, see [Configuration Properties](#cc-mysql-cdc-source-v2-config-properties). ### How do I filter which tables the connector captures? The MySQL CDC Source V2 connector filters tables and databases using regular expressions. 1. **Filter by database**: Use `database.include.list` or `database.exclude.list`: ```json { "database.include.list": "production,staging", "database.exclude.list": "test,development" } ``` 2. **Filter by table**: Use `table.include.list` or `table.exclude.list` with fully qualified names: ```json { "table.include.list": "production.orders,production.customers,staging.orders" } ``` 3. **Use regular expressions**: Filter tables with patterns: ```json { "table.include.list": "production\\.order_.*,production\\.customer_.*" } ``` Escape the dot character with a double backslash in JSON. 4. **Combine filters**: Include specific databases and exclude certain tables: ```json { "database.include.list": "production", "table.exclude.list": "production\\.temp_.*,production\\.backup_.*" } ``` #### IMPORTANT - Table names must be fully qualified: `database.table` - Use Java regular expression syntax - System tables such as `mysql.*` and `information_schema.*` are automatically excluded ### What does the error `The connector is trying to read binlog starting at` mean? This error means the connector’s last recorded position, a binlog file and position, or a GTID set, is no longer available on the MySQL server. This usually happens because the binlog files were purged, either by the server’s automatic purge or because the connector was offline longer than the binlog retention period. Error message: ```text The connector is trying to read binlog starting at GTIDSet '...' or binlog filename 'mysql-bin.000123', position 4, but this is no longer available on the server ``` Recover from the error: The connector cannot resume from a position that no longer exists. Increasing binlog retention does not recover an already-purged position. Recover using one of the following options: 1. **Use the** `when_needed` **snapshot mode**: When `snapshot.mode` is set to `when_needed`, the connector automatically takes a new snapshot when the recorded offset points to a binlog position or GTID that is no longer available on the server. This avoids manually editing offsets. As with the `initial` mode, this re-snapshots the included tables and can produce duplicate records for data that consumers already processed. 2. **Reset the connector offsets**: Delete the existing offsets so the connector starts from a valid position. For the procedure, see [Manage custom offsets](#cc-mysql-cdc-source-v2-debezium-custom-offsets). After you reset the offsets, the outcome depends on the configured `snapshot.mode`: * `initial`: the connector takes a new snapshot of all included tables and then resumes streaming. This can produce duplicate records for data that consumers already processed. * `no_data`: the connector snapshots only the schema and then streams from the current binlog position. No table data is re-snapshotted. #### IMPORTANT Do not use the `recovery` snapshot mode to recover from this error. That mode recovers a corrupted or lost database schema history topic for a connector that is already capturing changes, and it requires an existing offset. To prevent the error from recurring, set an adequate binlog retention period. Confluent recommends seven days. Set this in the MySQL server configuration file rather than at runtime: ```ini [mysqld] binlog_expire_logs_seconds = 604800 # 7 days ``` ### Why is my connector not capturing changes from specific tables? If a table’s changes are not appearing, the cause is usually table filtering or an in-progress snapshot. Check the following: 1. **Check table filtering**: `table.include.list` is a regular expression. The fully qualified table name, `database.table`, must match the pattern you specify: ```json { "table.include.list": "mydb\\.orders,mydb\\.customers" } ``` 2. **Verify table is not excluded**: Check that the table is not in `table.exclude.list` or matches an exclude pattern. 3. **Review connector logs**: Look for warnings or errors about specific tables in the Cloud Console. 4. **Confirm the snapshot completed**: During the initial snapshot, the connector does not stream changes until the snapshot of all included tables completes. If the snapshot is still in progress, changes do not appear for any table yet. ### How do I configure SSL/TLS encryption for MySQL connections? The MySQL CDC Source V2 connector supports SSL/TLS encryption for secure database connections. 1. **Enable SSL on MySQL server**: Verify SSL is enabled: ```sql SHOW VARIABLES LIKE 'have_ssl'; ``` 2. **Configure connector for SSL**: ```json { "database.ssl.mode": "required" } ``` SSL modes: * `disabled`: No SSL encryption * `preferred`: Use SSL if available and fall back to an unencrypted connection * `required`: Require SSL and fail if unavailable * `verify_ca`: Require SSL and verify the CA certificate * `verify_identity`: Require SSL and verify the server identity For details about each SSL mode, see MySQL’s [Connector/J security configuration properties](https://dev.mysql.com/doc/connector-j/en/connector-j-connp-props-security.html). 1. **For AWS RDS MySQL**: Set the SSL mode to `required` to encrypt the connection. To also verify the server certificate, use `verify_ca` or `verify_identity` and upload the RDS CA certificate as described in the next step. 2. **For custom CA certificates**: When using `verify_ca` or `verify_identity`, upload the truststore that contains the CA certificate using the **SSL Truststore** and **SSL Truststore Password** connector settings. 3. **Verify SSL connection**: Check MySQL logs or run: ```sql SHOW STATUS LIKE 'Ssl_cipher'; ``` ### How do I handle schema changes in my MySQL database? The MySQL CDC Source V2 connector handles many schema changes automatically. Other changes are breaking and require you to plan for the downstream impact. Supported schema changes: 1. **Adding columns**: New columns are automatically detected and included in change events: * **Nullable columns**: Appear with `null` until the column is populated. * **Columns with defaults**: Appear with their default value. 2. **Dropping columns**: Removed columns no longer appear in change events. 3. **Renaming columns**: Tracked natively through the connector’s schema history. Change events after the rename use the new column name. The connector does not emit a drop event followed by an add event. The connector automatically captures changes made through data definition language, or DDL. No additional configuration is required to handle supported schema changes. When using Schema Registry, the compatibility mode set on the subject (for example, `-value`) governs which schema changes are allowed. For the full compatibility matrix, see [Schema Evolution and Compatibility](https://docs.confluent.io/platform/current/schema-registry/fundamentals/schema-evolution.html). Breaking schema changes: * **Changing the primary key**: The connector continues running, but the Kafka message key schema changes mid-stream. For example, keys change from `{"id":5}` to `{"id":5,"product":"elderberry"}` after a single-column key becomes composite. This can break downstream consumers and Schema Registry compatibility. Plan for the key schema change before altering a primary key. * **Changing the column data type**: Can cause deserialization errors. * **Dropping a monitored table**: Remove the table from `table.include.list` before you drop it. Dropping one of several monitored tables does not fail the connector. If you drop the last table in `table.include.list`, the connector fails on the next task restart with the error `After applying the include/exclude list filters, no changes will be captured. Please check your configuration!` Best practices: 1. **Test schema changes**: Test DDL changes in a non-production environment first. 2. **Pause connector during major changes**: For breaking changes, pause the connector, make changes, then resume. 3. **Monitor for errors**: After schema changes, check connector logs for issues. ## Next Steps For an example that shows fully managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html). This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud. [![image](images/topology.png)](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)