Microsoft SQL Server CDC Source V2 (Debezium) Connector for Confluent Cloud¶
The fully-managed Microsoft SQL Server Change Data Capture (CDC) Source V2 (Debezium) connector for Confluent Cloud can obtain a snapshot of the existing data in a SQL Server database and then monitor and record all subsequent row-level changes to that data. The connector supports Avro, JSON Schema, Protobuf, or JSON (schemaless) output data formats. All of the events for each table are recorded in a separate Apache Kafka® topic. The events can then be easily consumed by applications and services.
Note
- This Quick Start is for version 2 of fully-managed cloud connector. If you are planning to upgrade from V1 to V2, see Moving from V1 to V2 Moving from V1 to V2.
- If you are installing the connector locally for Confluent Platform, see Debezium SQL Server CDC Source Connector for Confluent Platform. For more information, see Debezium 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.
V2 Improvements¶
Note the following improvements made to the V2 connector.
- Supports capturing changes from multiple databases on a single database engine. using a single connector instance and multiple tasks.
- Heartbeat events are emitted even if the connector does not find any changes or the changes that did occur are not of relevance to the connector.
- 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.
- 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 SQL Server CDC Source V2 (Debezium) connector provides the following features:
Topics created automatically: The connector automatically creates Kafka topics using the naming convention:
<topic.prefix>.<schemaName>.<tableName>
. The tables are created with the properties:topic.creation.default.partitions=1
andtopic.creation.default.replication.factor=3
. For more information, see Maximum message size.Note
For DB history topics, the connector automatically creates Kafka topics using the naming convention:
dbhistory.<topic.prefix>.<connect-id>
.Database authentication: Uses password authentication.
SSL support: Supports SSL encryption.
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 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 Kafka record key format. Schema Registry must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). See Schema Registry Enabled Environments for additional information.
Tasks per connector: The connector supports multiple tasks. More tasks may improve performance.
Incremental snapshot: Supports incremental snapshotting via signaling.
Offset management capabilities: Supports offset management. For more information, see Manage custom offsets.
For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Managed and Custom Connectors section.
Supported database versions¶
The SQL Server CDC Source V2 (Debezium) connector is compatible with the following SQL Server versions: 2017, 2019.
Limitations¶
Be sure to review the following information.
- For connector limitations, see Microsoft SQL Server CDC Source V2 (Debezium) Connector limitations.
- If you plan to use one or more Single Message Transforms (SMTs), see SMT Limitations.
- If you plan to use Confluent Cloud Schema Registry, see Schema Registry Enabled Environments.
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.
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 the change tables so that when the snapshot is complete, the CDC connector can start processing database changes that have completed since the snapshot process began.
If one or more of the tables are very large, the snapshot process could run longer than the retention period set for the cleanup job to purge the change tables. To capture very large tables, you should retain the change tables for longer than normal by increasing the retention period of change tables.
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.
To manage offsets:
- Manage offsets using Confluent Cloud APIs. For more information, see Cluster API reference.
To get the current offset, make a GET
request that specifies the environment, Kafka cluster, and connector name.
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.
{
"id": "lcc-example123",
"name": "{connector_name}",
"offsets": [
{
"partition": {
"database": "database_name",
"server": "server_name"
},
"offset": {
"change_lsn": "00000118:0000302b:0003",
"commit_lsn": "00000118:0000302b:0004",
"event_serial_no": 1,
"transaction_id": null
}
}
],
"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. Useobserved_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. CallingGET
repeatedly will fetch more recently observed offsets. - Information about the connector.
- In these examples, the curly braces around “{connector_name}” indicate a replaceable value.
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.
POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request
Host: https://api.confluent.cloud
{
"type": "PATCH",
"offsets": [
{
"partition": {
"database": "database_name",
"server": "server_name"
},
"offset": {
"change_lsn": "0000010d:0000138f:0002",
"commit_lsn": "00000116:000042CC:0004",
"event_serial_no": 1,
"transaction_id": null
}
}
]
}
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.
{
"id": "lcc-example123",
"name": "{connector_name}",
"offsets": [
{
"partition": {
"database": "database_name",
"server": "server_name"
},
"offset": {
"change_lsn": "0000010d:0000138f:0002",
"commit_lsn": "00000116:000042CC:0004",
"event_serial_no": 1,
"transaction_id": null
}
}
],
"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.
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.
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.
{
"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.
To get the status of a previous offset request, make a GET
request that specifies the environment, Kafka cluster, and connector
name.
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.
{
"request": {
"id": "lcc-example123",
"name": "{connector_name}",
"offsets": [
{
"partition": {
"database": "database_name",
"server": "server_name"
},
"offset": {
"change_lsn": "0000010d:0000138f:0002",
"commit_lsn": "00000116:000042CC:0004",
"event_serial_no": 1,
"transaction_id": null
}
}
],
"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": {
"database": "database_name",
"server": "server_name"
},
"offset": {
"change_lsn": "0000010d:0000138f:0002",
"commit_lsn": "00000116:000042CC:0004",
"event_serial_no": 1,
"transaction_id": null
}
},
],
"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 Microsoft SQL Server Change Data Capture (CDC) Source V2 (Debezium) connector.
Field | Definition | Required/Optional |
---|---|---|
database |
The Microsoft SQL Server database. | Required |
server |
The name of the Microsoft SQL Server server. | Required |
change_lsn |
change_lsn represents the Log Sequence Number (LSN) associated with the specific change event within the transaction. |
Required |
commit_lsn |
commit_lsn represents the Log Sequence Number (LSN) associated with the commit of the transaction that included the change event. |
Required |
event_serial_no |
For update events there can be duplicate values for change_lsn and commit_lsn . In such cases, event_serial_no differentiates
the old value (1 ) and new value (2 ) of the event. |
Required |
transaction_id |
If you set provide.transaction.metadata to true in the connector configuration, transaction_id contains
the transaction ID. Otherwise, transaction_id is null. |
Optional |
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 orschema.history.internal.kafka.topic
config value using the Confluent CLI. - To reuse schema history topic, the specified value of
Topic prefix
andDatabase names
should be same as that of the self-managed connector and the offsets to create connector with should be provided for all the databases.
Quick Start¶
Use this quick start to get up and running with the Confluent Cloud Microsoft SQL Server 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 Microsoft SQL Server database and then monitoring and recording all subsequent row-level changes.
- Prerequisites
Authorized access to a 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.
Schema Registry must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). See Schema Registry Enabled Environments for additional information.
SQL Server configured for change data capture (CDC).
- For Debezium instructions, see Setting up SQL Server.
- For Amazon RDS instructions, see Using change data capture.
Public access may be required for your database. See Manage Networking for Confluent Cloud Connectors for details. The example below shows the AWS Management Console when setting up a Microsoft SQL Server database.
For networking considerations, see Networking and DNS. To use a set of public egress IP addresses, see Public Egress IP Addresses for Confluent Cloud Connectors. The following example shows the AWS Management Console when setting up security group rules for the VPC.
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 resource ID.
- Create a Confluent Cloud service account for the connector. Make sure to review the ACL entries required in the service account documentation. 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 or you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.
Using the Confluent Cloud Console¶
Step 1: Launch your Confluent Cloud cluster¶
See the Quick Start for Confluent Cloud for installation instructions.
Step 2: Add a connector¶
In the left navigation menu, click Connectors. If you already have connectors in your cluster, click + Add connector.
Step 4: Enter the connector details¶
Note
- Make sure you have all your prerequisites completed.
At the Microsoft SQL Server CDC Source V2 (Debezium) screen, complete the following:
- 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. 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.
- Click Continue.
- Add the following database connection details:
- Database hostname: IP address or hostname of the Microsoft SQL Server database server.
- Database port: Port number of the Microsoft SQL Server database server.
- Database username: The name of the Microsoft SQL Server user that has the required authorization.
- Database password: The password of the Microsoft SQL Server user that has the required authorization.
- Database names: Comma separated list of names of the Microsoft SQL Server databases from which to stream the changes.
- Database encrypt: Controls SSL encryption for connections to a SQL Server database. The default value is false, indicating that the connector won’t force the server to support TLS encryption. When set to true, the connector requests to use TLS encryption with the server.
- Click Continue.
Add the following details:
- Select the output record value format (data going to the Kafka topic): AVRO, JSON, JSON_SR (JSON Schema), or PROTOBUF. Schema Registry must be enabled to use a Schema Registry-based format (for example, Avro, JSON Schema, or Protobuf). For additional information, see Schema Registry Enabled Environments.
- Topic prefix: Provides a namespace (logical server name) for the SQL Server database server from which Debezium is capturing changes. This logical name forms a namespace and is used in all the names of the Kafka topics and the Kafka Connect schema names.
- Snapshot mode: Specifies the criteria for running a snapshot
when the connector starts. Possible settings are:
initial
,initial_only
, andschema_only
.initial
(default): Takes a snapshot of structure and data of captured tables; useful if topics should be populated with a complete representation of the data from the captured tables.initial_only
: Takes a snapshot of structure and data likeinitial
but instead does not transition into streaming changes once the snapshot has completed.schema_only
: Takes a snapshot of the structure of captured tables only; useful if only changes happening from now onwards should be propagated to topics.
- 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
schemaName.tableName
. This property cannot be used with the property 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
schemaName.tableName
. This property cannot be used with the property Tables included.
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?.
JSON output 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 andNUMERIC
to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.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
.Database schema history topic name: 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.<topic-prefix>.<lcc-id>
.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.schemaName.tableName
. These signals can be used to perform incremental snapshotting.Specify application intent: Defines the connection property
applicationIntent
within the connection string. Possible settings are:ReadWrite
andReadOnly
. Defaults toReadWrite
.Snapshot isolation mode: A mode to control which transaction isolation level is used and how long the connector locks tables that are designated for capture. Possible settings are:
read_uncommitted
,read_committed
,repeatable_read
,snapshot
, andexclusive
. Thesnapshot
,read_committed
andread_uncommitted
modes do not prevent other transactions from updating table rows during initial snapshot. Theexclusive
andrepeatable_read
modes do prevent concurrent updates. Mode choice also affects data consistency. Onlyexclusive
andsnapshot
modes guarantee full consistency, that is, initial snapshot and streaming logs constitute a linear history. In case ofrepeatable_read
andread_committed
modes, it might happen that, for instance, a record added appears twice - once in initial snapshot and once in streaming phase. Nonetheless, that consistency level should do for data mirroring. Forread_uncommitted
there are no data consistency guarantees at all (some data might be lost or corrupted).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
schemaName.tableName.columnName
.Event processing failure handling mode: Specifies how the connector should react to exceptions during processing of events. Possible settings are:
fail
,skip
, andwarn
.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
, andavro_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
, andavro_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.
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 tofalse
. Consider setting this totrue
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.
Decimal handling mode: Specifies how the connector should handle values for
DECIMAL
andNUMERIC
columns. Possible settings are:precise
,double
, andstring
.precise
(default): represents values by usingjava.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
(default): captures the time and timestamp values exactly as in the database using either millisecond, microsecond, or nanosecond precision values based on the database column’s type.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 and Predicates: For details, see the Single Message Transforms (SMT) documentation. For additional information about the Debezium SMTs ExtractNewRecordState and EventRouter (Debezium), see Debezium transformations.
For all property values and definitions, see Configuration Properties.
Click Continue.
Based on the number of topic partitions you select, you will be provided with a recommended number of tasks.
- To change the number of tasks, use the Range Slider to select the desired number of tasks.
- Click Continue.
Verify the connection details by previewing the running configuration.
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.<topic.prefix>.<connect-id>
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 Managed and Custom Connectors 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 completed.
Step 1: List the available connectors¶
Enter the following command to list available connectors:
confluent connect plugin list
Step 2: List the connector configuration properties¶
Enter the following command to show the connector configuration properties:
confluent connect plugin describe <connector-plugin-name>
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 example shows the required connector properties.
{
"connector.class": "SqlServerCdcSourceV2",
"name": "SqlServerCdcSourceV2Connector_0",
"kafka.auth.mode": "KAFKA_API_KEY",
"kafka.api.key": "****************",
"kafka.api.secret": "****************************************************************",
"database.hostname": "connect-sqlserver-cdc.<host-id>.us-west-2.rds.amazonaws.com",
"database.port": "1433",
"database.user": "admin",
"database.password": "************",
"database.names": "testDb",
"topic.prefix": "sql",
"table.include.list":"dbo.passengers",
"output.data.format": "JSON",
"tasks.max": "1"
}
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
orKAFKA_API_KEY
(the default). To use an API key and secret, specify the configuration propertieskafka.api.key
andkafka.api.secret
, as shown in the example configuration (above). To use a service account, specify the Resource ID in the propertykafka.service.account.id=<service-account-resource-ID>
. To list the available service account resource IDs, use the following command:confluent iam service-account list
For example:
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 Microsoft SQL Server database server."database.port"
: Port number of the Microsoft SQL Server database server."database.user"
: The name of the Microsoft SQL Server user that has the required authorization."database.password"
: Password of the Microsoft SQL Server user that has the required authorization."database.names"
: The names of the Microsoft SQL Server databases from which to stream the changes."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 formschemaName.tableName
. This property cannot be used with the propertytable.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 in use by the connector.
Single Message Transforms: See the Single Message Transforms (SMT) documentation for details about adding SMTs using the CLI. For additional information about the Debezium SMTs ExtractNewRecordState and EventRouter (Debezium), see Debezium transformations.
See Configuration Properties for all property values and definitions.
Step 4: Load the properties file and create the connector¶
Enter the following command to load the configuration and start the connector:
confluent connect cluster create --config-file <file-name>.json
For example:
confluent connect cluster create --config-file microsoft-sql-cdc-source-v2.json
Example output:
Created connector SqlServerCdcSourceV2Connector_0 lcc-ix4dl
Step 5: Check the connector status¶
Enter the following command to check the connector status:
confluent connect cluster list
Example output:
ID | Name | Status | Type
+-----------+----------------------------------+---------+-------+
lcc-ix4dl | SqlServerCdcSourceV2Connector_0 | RUNNING | source
Step 6: Check the Kafka topic.¶
After the connector is running, verify that messages are populating your Kafka topic.
For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Managed and Custom Connectors section.
Note
A topic named dbhistory.<topic.prefix>.<connect-id>
is
automatically created for schema.history.internal.kafka.topic
with one partition.
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. To understand these changes and to plan for moving to version 2, see Backward Incompatible Changes in Debezium CDC V2 Connectors.
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. To get the offset in the following section, use the Confluent Cloud APIs. For more information, see Cluster API reference, Manage custom offsets, and Manage Offsets for Fully-Managed Connectors in Confluent Cloud.
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.
Pause the v1 connector.
Get the offset for the v1 connector.
Create the v2 connector using the offset from the previous step.
confluent connect cluster create [flags]
For example:
Create a configuration file with connector configs and offsets.
{ "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.
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.
Delete the v1 connector.
For more information, see Manage Offsets for Fully-Managed Connectors in Confluent Cloud.
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.
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.
- Type: string
- Default: KAFKA_API_KEY
- Valid Values: KAFKA_API_KEY, SERVICE_ACCOUNT
- 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
How should we connect to your database?¶
database.hostname
IP address or hostname of the SQL Server database server.
- Type: string
- Importance: high
database.port
Port number of the SQL Server database server.
- Type: int
- Valid Values: [0,…,65535]
- Importance: high
database.user
The name of the SQL Server database user that has the required authorization.
- Type: string
- Importance: high
database.password
The password for the SQL Server database user that has the required authorization.
- Type: password
- Importance: high
database.names
The comma-separated list of the SQL Server database names from which to stream the changes.
- Type: list
- Importance: high
database.encrypt
Controls SSL encryption for connections to a SQL Server database. The default value is false, indicating that the connector won’t force the server to support TLS encryption. When set to true, the connector requests to use TLS encryption with the server.
- Type: boolean
- Default: false
- Importance: high
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
- 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
json.output.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
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 tombstone event should be generated after a delete event.
true - a delete operation is represented by a delete event and a subsequent tombstone event.
false - only a delete event is emitted.
After a source record is deleted, emitting the tombstone event (the default behavior) allows Kafka to completely delete all events that pertain to the key of the deleted row in case log compaction is enabled for the topic.
- Type: boolean
- Default: true
- Importance: medium
How should we name your topic(s)?¶
topic.prefix
Topic prefix that provides a namespace (logical server name) for the particular SQL Server 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: <topic.prefix>.<databaseName>.<schemaName>.<tableName>.
- 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.<topic-prefix>.<lcc-id>
.- Type: string
- Default: dbhistory.${topic.prefix}.{{.logicalClusterId}}
- 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.schemaName.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, and schema_only.
initial - Takes a snapshot of structure and data of captured tables; useful if topics should be populated with a complete representation of the data from the captured tables.
initial_only - Takes a snapshot of structure and data like initial but instead does not transition into streaming changes once the snapshot has completed.
schema_only - Takes a snapshot of the structure of captured tables only; useful if only changes happening from now onwards should be propagated to topics.
- Type: string
- Default: initial
- Valid Values: initial, initial_only, schema_only
- Importance: medium
driver.applicationIntent
Defines the connection property applicationIntent within the connection string. Possible settings are: ReadWrite and ReadOnly.
- Type: string
- Default: ReadWrite
- Valid Values: ReadOnly, ReadWrite
- Importance: medium
snapshot.isolation.mode
A mode to control which transaction isolation level is used and how long the connector locks tables that are designated for capture. Possible settings are: read_uncommitted, read_committed, repeatable_read, snapshot, and exclusive. The snapshot, read_committed and read_uncommitted modes do not prevent other transactions from updating table rows during initial snapshot. The exclusive and repeatable_read modes do prevent concurrent updates. Mode choice also affects data consistency. Only exclusive and snapshot modes guarantee full consistency, that is, initial snapshot and streaming logs constitute a linear history. In case of repeatable_read and read_committed modes, it might happen that, for instance, a record added appears twice - once in initial snapshot and once in streaming phase. Nonetheless, that consistency level should do for data mirroring. For read_uncommitted there are no data consistency guarantees at all (some data might be lost or corrupted).
- Type: string
- Default: repeatable_read
- Valid Values: exclusive, read_committed, read_uncommitted, repeatable_read, snapshot
- Importance: low
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 schemaName.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 schemaName.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
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 schemaName.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
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
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 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.
- 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. Possible settings are: none, avro, and avro_unicode.
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 like _uxxxx. Note: _ 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
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
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
value.converter.ignore.default.for.nullables
Specifies whether the default value should be ignored for nullable fields. If set to true, the default value is ignored for nullable fields. If set to false, the default value is used for nullable fields.
- Type: boolean
- Default: false
- 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
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: high
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 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 captures the time and timestamp values exactly as in the database using either millisecond, microsecond, or nanosecond precision values based on the database column’s type.
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
- Valid Values: adaptive, connect
- Importance: medium
Number of tasks for this connector¶
tasks.max
Maximum number of tasks that the connector can use to capture data from the database instance. If the database.names list contains more than one element, you can increase the value of this property to a number less than or equal to the number of elements in the list.
- Type: int
- Valid Values: [1,…,10]
- Importance: high
Next Steps¶
For an example that shows fully-managed Confluent Cloud connectors in action with Confluent Cloud ksqlDB, see the Cloud ETL Demo. This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud.