anchor.diff:  
   // Type: assembly
   // ModuleID: setting-up-sql-server-for-use-with-the-debezium-sql-server-connector
  -//Title: Setting up SQL Server for use with the {prodname} connector
  +//Title: Setting up SQL Server to run a {prodname} connector
   [[setting-up-sqlserver]]
   == Setting up SQL Server
   
  
positive.code: 
  endif::community[]
  
  ifdef::product[]
  
  The following example shows the configuration for a connector instance with the logical name `fullfillment` that captures a SQL Server server at port 1433 on 192.168.99.100.
  Typically, you configure the {prodname} SQL Server connector in a `.yaml` file in which you assign values to the configuration properties that are available for the connector.
  
  [source,yaml,options="nowrap"]
  ----
  apiVersion: kafka.strimzi.io/v1beta1
    kind: KafkaConnector
    metadata:
      name: inventory-connector // <1>
      labels: strimzi.io/cluster: my-connect-cluster
    spec:
      class: io.debezium.connector.sqlserver.SqlServerConnector // <2>
      config:
        database.hostname: 192.168.99.100 // <3>
        database.port: 1433 // <4>
        database.user: debezium // <5>
        database.password: dbz // <6>
        database.dbname: testDB // <7>
        database.server.name: fullfullment // <8>
        database.include.list: dbo.customers // <9>
        database.history.kafka.bootstrap.servers: my-cluster-kafka-bootstrap:9092 // <10>
        database.history.kafka.topic: dbhistory.fullfillment // <11>
  
  ----
  
  .Descriptions of connector configuration settings
  [cols="1,7",options="header",subs="+attributes"]
  |===
  |Item |Description
  
  |1
  |The name of our connector when we register it with a Kafka Connect service.
  
  |2
  |The name of this SQL Server connector class.
  
  |3
  |The address of the SQL Server instance.
  
  |4
  |The port number of the SQL Server instance.
  
  |5
  |The name of the SQL Server user.
  
  |6
  |The password for the SQL Server user.
  
  |7
  |The name of the database to capture changes from.
  
  |8
  |The logical name of the SQL Server instance/cluster, which forms a namespace and is used in all the names of the Kafka topics to which the connector writes, the Kafka Connect schema names, and the namespaces of the corresponding Avro schema when the Avro converter is used.
  
  |9
  |A list of all tables whose changes {prodname} should capture.
  
  |10
  |The list of Kafka brokers that this connector will use to write and recover DDL statements to the database history topic.
  
  |11
  |The name of the database history topic where the connector will write and recover DDL statements. This topic is for internal use only and should not be used by consumers.
  
  |===
  
  endif::product[]
  
  See the {link-prefix}:{link-sqlserver-connector}#sqlserver-connector-properties[complete list of connector properties] that can be specified in these configurations.
  
  This configuration can be sent via POST to a running Kafka Connect service, which will then record the configuration and start up the one connector task that will connect to the SQL Server database, read the transaction log, and record events to Kafka topics.
  
  // Type: procedure
  // ModuleID: adding-sql-server-connector-configuration-to-kafka-connect
  //Title: Adding SQL Server connector configuration to Kafka Connect
  [[sqlserver-adding-connector-configuration]]
  ==== Adding connector configuration
  
  ifdef::community[]
  To run a {prodname} SQL Server connector, create a connector configuration and add the configuration to your Kafka Connect cluster.
  
  .Prerequisites
  
  * SQL Server is set up to run a {prodname} connector.
  
  * A {prodname} SQL Server connector is installed.
  
  .Procedure
  
  . Create a configuration for the SQL Server connector.
  
  . Use the link:{link-kafka-docs}/#connect_rest[Kafka Connect REST API] to add that connector configuration to your Kafka Connect cluster.
  
  endif::community[]
  
  ifdef::product[]
  
  You can use a provided {prodname} container to deploy a {prodname} SQL Server connector.
  In this procedure, you build a custom Kafka Connect container image for {prodname}, configure the {prodname} connector as needed,
  and then add your connector configuration to your Kafka Connect environment.
  
  .Prerequisites
  
  * Podman or Docker is installed and you have sufficient rights to create and manage containers.
  * You installed the {prodname} SQL Server connector archive.
  
  .Procedure
  
  . Extract the {prodname} SQL Server connector archive to create a directory structure for the connector plug-in, for example:
  +
  [subs=+macros]
  ----
  pass:quotes[*tree ./my-plugins/*]
  ./my-plugins/
  ├── debezium-connector-sqlserver
  │   ├── ...
  ----
  
  . Create and publish a custom image for running your {prodname} connector:
  
  .. Create a new `Dockerfile` by using `{DockerKafkaConnect}` as the base image. In the following example, you would replace _my-plugins_ with the name of your plug-ins directory:
  +
  [subs="+macros,+attributes"]
  ----
  FROM {DockerKafkaConnect}
  USER root:root
  pass:quotes[COPY _./my-plugins/_ /opt/kafka/plugins/]
  USER 1001
  ----
  +
  Before Kafka Connect starts running the connector, Kafka Connect loads any third-party plug-ins that are in the `/opt/kafka/plugins` directory.
  
  .. Build the container image. For example, if you saved the `Dockerfile` that you created in the previous step as `debezium-container-for-sqlserver`, and if the `Dockerfile` is in the current directory, then you would run the following command:
  +
  `podman build -t debezium-container-for-sqlserver:latest .`
  
  .. Push your custom image to your container registry, for example:
  +
  `podman push debezium-container-for-sqlserver:latest`
  
  .. Point to the new container image. Do one of the following:
  +
  * Edit the `spec.image` property of the `KafkaConnector` custom resource. If this property is set, it overrides the `STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE` variable in the Cluster Operator. For example:
  +
  [source,yaml,subs="+attributes"]
  ----
  apiVersion: {KafkaConnectApiVersion}
  kind: KafkaConnector
  metadata:
    name: my-connect-cluster
  spec:
    #...
    image: debezium-container-for-sqlserver
  ----
  +
  * In the `install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml` file, edit the `STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE` variable to point to the new container image and reinstall the Cluster Operator. If you edit this file you must apply it to your OpenShift cluster.
  
  . Create a `KafkaConnector` custom resource that defines your {prodname} SQL Server connector instance. See {LinkDebeziumUserGuide}#sqlserver-example-configuration[the connector configuration example].
  
  . Apply the connector instance, for example:
  +
  `oc apply -f inventory-connector.yaml`
  +
  This registers `inventory-connector` and the connector starts to run against the `inventory` database.
  
  . Verify that the connector was created and has started to capture changes in the specified database. You can verify the connector instance by watching the Kafka Connect log output as, for example, `inventory-connector` starts.
  
  .. Display the Kafka Connect log output:
  +
  [source,shell,options="nowrap"]
  ----
  oc logs $(oc get pods -o name -l strimzi.io/name=my-connect-cluster-connect)
  ----
  
  .. Review the log output to verify that the initial snapshot has been executed. You should see something like the following lines:
  +
  [source,shell,options="nowrap"]
  ----
  ... INFO Starting snapshot for ...
  ... INFO Snapshot is using user 'debezium' ...
  ----
  
  endif::product[]
  
  .Results
  
  When the connector starts, it {link-prefix}:{link-sqlserver-connector}#sqlserver-snapshots[performs a consistent snapshot] of the SQL Server databases that the connector is configured for. The connector then starts generating data change events for row-level operations and streaming change event records to Kafka topics.
  
  
  // Type: reference
  // ModuleID: descriptions-of-debezium-sql-server-connector-configuration-properties
  // Title: Descriptions of {prodname} SQL Server connector configuration properties
  [[sqlserver-connector-properties]]
  ==== Connector properties
  
  The following configuration properties are _required_ unless a default value is available.
  
  
negative.code: 
  For details about how the connector works, see the following sections:
  
  * xref:sqlserver-snapshots[]
  * xref:how-the-debezium-sql-server-connector-reads-change-data-tables[]
  * xref:default-names-of-kafka-topics-that-receive-debezium-sql-server-change-event-records[]
  * xref:how-the-debezium-sql-server-connector-uses-the-schema-change-topic[]
  