<a id="sqlserver-source-connector"></a>

#### IMPORTANT
Confluent only provides support for specific versions of Debezium-built connectors on Confluent Platform. For complete details on the latest supported Debezium 3.1.2 version, see the
official [Debezium SQL Server Source Connector](https://debezium.io/documentation/reference/3.1/connectors/sqlserver.html) documentation.

# Debezium SQL Server Source Connector for Confluent Platform

The [Debezium SQL Server Source connector](https://debezium.io/docs/connectors/sqlserver/) is a connector that can take
a snapshot of the existing data in a SQL Server database and then monitor and
record all subsequent row-level changes to that data. All of the events for each
table are recorded in a separate Apache Kafka® topic, where they can be easily
consumed by applications and services.

## Features

The Debezium SQL Server Source connector includes the following features:

- [At least once delivery](#debezium-mysql-server-source-at-least-once-delivery)
- [Supports one task](#debezium-mysql-server-source-supports-one-task)
- [Automatic topic creation](#debezium-mysql-server-source-auto-topic-creation)
- [Client-side encryption](#debezium-mysql-server-source-sink-csfle-sm)

<a id="debezium-mysql-server-source-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records are delivered at least once to the Kafka
topic. If the connector restarts, there may be some duplicate
records in the Kafka topic.

<a id="debezium-mysql-server-source-supports-one-task"></a>

### Supports one task

The Debezium SQL Server Source connector supports running only one task.

<a id="debezium-mysql-server-source-auto-topic-creation"></a>

### Automatic topic creation

The connector automatically creates the internal database history Kafka topic if
it doesn’t exist.

<a id="debezium-mysql-server-source-sink-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

### Kerberos Integrated Authentication

The connector supports Kerberos integrated authentication. Follow the steps below to set up the connector with Kerberos integrated authentication:

1. Set up a managed Microsoft AD instance on AWS (for example, directory name: cdc.example.com). For more information, see [AWS RDS for SQL server](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_SQLServerWinAuth.html#USER_SQLServerWinAuth.SettingUp).
2. Create a managed SQL Server RDS instance.
3. Enable Windows Authentication using the AD instance’s domain identifier.

   #### NOTE
   The directory and database instance should both provisioned in the same VPC.
4. Create a Windows login for the AD user using the following SQL command:
   ```none
   CREATE LOGIN [CDC\admin] FROM WINDOWS WITH DEFAULT_DATABASE = [master], DEFAULT_LANGUAGE = [us_english];
   ```
5. Launch an EC2 instance (for example, Amazon Linux 2023) and connect it to the same AD directory. To join the EC2 instance to the directory, see [Amazon EC2 Linux instance](https://docs.aws.amazon.com/directoryservice/latest/admin-guide/join_linux_instance.html).
6. Set the following JDBC connection properties (For more information, see [Connect to SQL server](https://learn.microsoft.com/en-us/sql/connect/jdbc/using-kerberos-integrated-authentication-to-connect-to-sql-server?view=sql-server-ver15).):
   ```none
   integratedSecurity=true
   authenticationScheme=JavaKerberos
   ```
7. Validate Kerberos Authentication using a sample JDBC program:
   ```none
    public static void main(String[] args) throws Exception {
    String connectionUrl =
      "jdbc:sqlserver://url-db-sqlserver.cdc.example.com:1433;"
      + "database=master;"
      + "integratedSecurity=true;"
      + "authenticationScheme=JavaKerberos;"
      + "encrypt=false;";
    ResultSet resultSet = null;

    try (Connection connection = DriverManager.getConnection(connectionUrl);
      Statement statement = connection.createStatement();) {
      String selectSql = "select auth_scheme from sys.dm_exec_connections where session_id=@@spid";
      resultSet = statement.executeQuery(selectSql);
      while (resultSet.next()) {
          System.out.println(resultSet.getString(1));
      }
     }
   }
   ```

   #### NOTE
   This output should print KERBEROS if authentication is successful.
8. Ensure the Debezium SQL Server connector configuration includes the following:
   ```none
   driver.integratedSecurity=true
   driver. authenticationScheme=JavaKerberos
   ```
9. Omit `database.user` and `database.password` as they are not required for Kerberos. Use the driver.\* prefix to pass these properties to the SQL Server JDBC driver used by Debezium.
10. Launch the Debezium SQL Server CDC Source connector with the configured settings.

## Install the SQL Server Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install debezium/debezium-connector-sqlserver:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install debezium/debezium-connector-sqlserver:<version-number>
  ```
- The Debezium PostgreSQL Source connector has specific ACL requirements. See
  the [ACL requirements for Debezium Source
  connectors](/cloud/current/connectors/service-account.html#additional-acl-entries) to ensure
  you meet the specified requirements.

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/debezium/debezium-connector-sqlserver/) for your
connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## License

The Debezium SQL Server Source connector is an open source connector and does not require a Confluent Enterprise License.

## Linux on IBM Z (s390x) support

Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x).
The connector supports the same capability available on x86_64 unless
otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support).

## Configuration Properties

For a complete list of configuration properties for this connector, see [Configuration Reference for Debezium SQL Server Source Connector for Confluent Platform](sqlserver_source_connector_config.md#sqlserver-source-connector-config).

### Configure Change Data Capture on SQL Server

The SQL Server database must be configured to enable the [Change Data Capture](https://docs.microsoft.com/en-us/sql/relational-databases/track-changes/enable-and-disable-change-data-capture-sql-server?view=sql-server-2017/)
(CDC) feature. The connector requires this feature be enabled for the table(s)
that should be captured. The functionality of the connector is based upon this
CDC feature which is included in the SQL Server Standard and SQL Server Enterprise
editions.

To enable CDC on the monitored database, use the following SQL command:

```none
USE MyDB
GO
EXEC sys.sp_cdc_enable_db
GO
```

Enable CDC for each table that you plan to monitor:

```none
USE MyDB
GO
EXEC sys.sp_cdc_enable_table @source_schema = N’dbo’, @source_name = N’MyTable’, @role_name = N’MyRole’, @filegroup_name = N’MyDB_CT’, @supports_net_changes = 1
GO
```

<a id="sqlserver-quickstart"></a>

## Quick Start

[Debezium’s SQL Server Source connector](https://debezium.io/docs/connectors/sqlserver/)  is a source connector that
can record events for each table in a separate Kafka topic, where they can be
easily consumed by applications and services.

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

### Install the Connector

If you want to use Docker images for setting up Kafka, ZooKeeper and Connect, refer
to the [Debezium tutorial](https://github.com/debezium/debezium-examples/tree/master/tutorial#using-sql-server/).
For the following tutorial, it is required to have a local setup of the Confluent Platform.
Note that as of Confluent Platform 7.5, ZooKeeper is deprecated for new deployments. Confluent
recommends KRaft mode for new deployments.

Navigate to your Confluent Platform installation directory and run the following command to
install the connector:

```bash
confluent connect plugin install debezium/debezium-connector-sqlserver:latest
```

Adding a new connector plugin requires restarting Connect. Use the
Confluent CLI to restart Connect.

```bash
confluent local services connect stop && confluent local services connect start
Using CONFLUENT_CURRENT: /Users/username/Sandbox/confluent-snapshots/var/confluent.NuZHxXfq
Starting Zookeeper
Zookeeper is [UP]
Starting Kafka
Kafka is [UP]
Starting Schema Registry
Schema Registry is [UP]
Starting Kafka REST
Kafka REST is [UP]
Starting Connect
Connect is [UP]
```

Check if the SQL Server plugin has been installed correctly and picked
up by the plugin loader.

```bash
curl -sS localhost:8083/connector-plugins | jq '.[].class' | grep SqlServer
"io.debezium.connector.sqlserver.SqlServerConnector"
```

### Set up SQL Server using Docker (Optional)

If you do not have a native installation of SQL Server, you may use the following command to bring up SQL Server with a Docker image.

```bash
#Pull docker image
docker pull mcr.microsoft.com/mssql/server:2017-latest

#Run docker container
docker run -e 'ACCEPT_EULA=Y' -e 'MSSQL_AGENT_ENABLED=true' \
-e 'MSSQL_PID=Standard' -e 'SA_PASSWORD=Password!' \
-p 1433:1433 --name sqlserver_1 \
-d mcr.microsoft.com/mssql/server:2017-latest

#Log into container to get your SQL Server command prompt
docker exec -it sqlserver_1 bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P Password!'
```

### Create Test Data and Enable Change Data Capture

The database operator must enable [Change Data Capture](https://docs.microsoft.com/en-us/sql/relational-databases/track-changes/enable-and-disable-change-data-capture-sql-server?view=sql-server-2017/) (CDC) for the table(s) that should be captured by the Debezium connector.
The functionality of the connector is based upon this CDC feature included in
the SQL Server Standard (beginning with SQL Server 2016 SP1) and SQL Server
Enterprise editions.

To enable CDC on the monitored database, use the following SQL command:

```none
USE MyDB
GO
EXEC sys.sp_cdc_enable_db
GO
```

Enable CDC for each table that you plan to monitor.

```none
USE MyDB
GO
EXEC sys.sp_cdc_enable_table @source_schema = N'dbo', @source_name = N'MyTable', @role_name = N'MyRole', @filegroup_name = N'MyDB_CT', @supports_net_changes = 1
GO
```

In this example, the database testDB is populated with a set of customer records.

Create `inventory.sql` with the following list of commands.

```none

-- Create the test database
CREATE DATABASE testDB;
GO
USE testDB;
EXEC sys.sp_cdc_enable_db;

-- Create some customers ...
CREATE TABLE customers (
  id INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE
);
INSERT INTO customers(first_name,last_name,email)
  VALUES ('Sally','Thomas','sally.thomas@acme.com');
INSERT INTO customers(first_name,last_name,email)
  VALUES ('George','Bailey','gbailey@foobar.com');
INSERT INTO customers(first_name,last_name,email)
  VALUES ('Edward','Walker','ed@walker.com');
INSERT INTO customers(first_name,last_name,email)
  VALUES ('Anne','Kretchmar','annek@noanswer.org');
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'customers', @role_name = NULL, @supports_net_changes = 0;
GO
```

Change Data Capture is enabled on the customers table and testDB database here.

To execute `inventory.sql` in the Docker container’s `sqlcmd` prompt, use the following command:

```bash
#Load inventory.sql through your container's sqlcmd prompt
cat inventory.sql | docker exec -i sqlserver_1 bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P Password!'
```

To execute `inventory.sql` on your native installation, use the following command:

```bash
sqlcmd -S myServer\instanceName -i C:\inventory.sql
```

### Start the Debezium SQL Server connector

#### Standalone cluster

1. Create the file `register-sqlserver.properties` to store the following connector configuration:
   ```text
   name=inventory-connector
   connector.class=io.debezium.connector.sqlserver.SqlServerConnector
   tasks.max=1
   database.server.name=server1
   database.hostname=localhost
   database.port=1433
   database.user=sa
   database.password=Password!
   database.dbname=testDB
   database.history.kafka.bootstrap.servers=localhost:9092
   database.history.kafka.topic=schema-changes.inventory
   ```
2. Start the connector:
   ```bash
   ./bin/connect-standalone  ./etc/.properties ./etc/register-sqlserver.properties
   ```

#### Distributed cluster

1. Create the file `register-sqlserver.json` to store the following connector configuration:
   ```none
   {
    "name": "inventory-connector",
    "config": {
        "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
        "tasks.max" : "1",
        "database.server.name" : "server1",
        "database.hostname" : "localhost",
        "database.port" : "1433",
        "database.user" : "sa",
        "database.password" : "Password!",
        "database.dbname" : "testDB",
        "database.history.kafka.bootstrap.servers" : "localhost:9092",
        "database.history.kafka.topic": "schema-changes.inventory"
        }
    }
   ```
2. Start the connector.
   ```none
   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-sqlserver.json
   ```

### Start your Kafka consumer

Start the consumer in a new terminal session.

```bash
confluent local consume server1.dbo.customers --from-beginning
```

When you enter SQL queries in your SQL Server bash, to add or modify records in the database, messages populate and are displayed on your consumer terminal to reflect those records.

```bash
USE testDB;
INSERT INTO customers(first_name,last_name,email) VALUES ('Pam','Thomas','pam@office.com');
GO
```

### Clean up resources

Delete the connector and stop Confluent services.

```bash
curl -X DELETE localhost:8083/connectors/inventory-connector
confluent local stop
```

Stop SQL Server container:

```bash
docker stop sqlserver_1
```

#### NOTE
Portions of the information provided here derives from documentation originally
produced by the [Debezium Community](https://debezium.io/). Work produced
by Debezium is licensed under [Creative Commons 3.0](https://creativecommons.org/licenses/by/3.0/).