working connect-sql connector

Author: mccullya

examples/custom-connect-sql/README.md

@@ -1,65 +1,59 @@
-As the README describes, you can reuse the Docker daemon from Minikube with eval $(minikube docker-env).
-
-So to use an image without uploading it, you can follow these steps:
-
-Set the environment variables with eval $(minikube docker-env)
-Build the image with the Docker daemon of Minikube (eg docker build -t my-image .)
-Set the image in the pod spec like the build tag (eg my-image)
-Set the imagePullPolicy to Never, otherwise Kubernetes will try to download the image.
-Important note: You have to run eval $(minikube docker-env) on each terminal you want to use, since it only sets the environment variables for the current shell session
-
-
-
-
-/opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "nbBg8G4DkR83Xs"
-
-select name from sys.databases
-go
-
-/opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "nbBg8G4DkR83Xs"
-USE AdventureWorks
-GO  
-EXEC sys.sp_cdc_enable_db  
-GO
-
-Deploying a connector:
-
-curl -X POST -H "Content-Type: application/json" --data @config.json http://localhost:8083/connectors
-
-Connect REST API Docs:
-https://docs.confluent.io/platform/current/connect/references/restapi.html
-
-
-curl -X POST -H "Content-Type: application/json" https://localhost:8083/connectors
-
-
-https://debezium.io/documentation/reference/1.0/connectors/sqlserver.html
-
-
--- ====
--- Enable Database for CDC template
--- ====
-USE MyDB
-GO
-EXEC sys.sp_cdc_enable_db
-GO
-
--- =========
--- Enable a Table Specifying Filegroup Option Template
--- =========
-USE MyDB
-GO
-
-EXEC sys.sp_cdc_enable_table
-@source_schema = N'dbo',
-@source_name   = N'Person',
-@role_name     = N'MyRole',
-@filegroup_name = N'MyDB_CT',
-@supports_net_changes = 0
-GO
-
-sqlcmd -S myServer\instanceName -i C:\myScript.sql
-
-
-
-
+## Custom Connect Image / Debezium SQL Server
+In this example we go through the following process:
+
+* Build a 'custom' kafka connect image with [Debezium](https://debezium.io/) plug-in, and make available to the internal (minikube) kubernetes cluster
+* Initiate a SQLServer stub populated with the traditional '[AdventureWorks](https://docs.microsoft.com/en-us/sql/samples/adventureworks-install-configure?view=sql-server-ver15&tabs=ssms)' database, and enable [CDC](https://en.wikipedia.org/wiki/Change_data_capture) on select tables
+* Deploy a connector via a cURL command
+* Observe how changes to CDC enabled tables will trigger events in Kafka
+
+NOTE: For ease of readability, we will simply reference the scripts that perform the actions of the following stages.  For better understanding of what is actually being done, please review the scripts themselves which will have their own comments/notations.  **Assumptions are that you will be running all commands from the present directory**
+
+### Building the custom docker image
+The Dockerfile installs a custom plugin with the following line: `RUN confluent-hub install --no-prompt debezium/debezium-connector-sqlserver:1.6.0`.  To build, run:
+
+```shell
+./build-inside.sh
+```
+### Deploy CRDs
+Deploy the CRDS using the standard way:
+```shell
+kubectl apply -k ../../kustomize/crds
+```
+### Deploy Confluent Operator and Confluent Services 
+Deploy the confluent operator and services:
+```shell
+kubectl apply -k .
+```
+### Enable CDC on 'person' table of AdventureWorks Database
+CDC needs to be enabled on a table by table basis.  This table is also referenced in the prod-mssql-connnector.json file. 
+```shell
+./enable_cdc.sh
+Context "minikube" modified.
+Changed database context to 'AdventureWorks'.
+Job 'cdc.AdventureWorks_capture' started successfully.
+Job 'cdc.AdventureWorks_cleanup' started successfully.
+```
+### Deploy Debezium Connector
+A curl request is sent to the 'connect pod' to install the connector.
+```shell
+./deploy_connector.sh
+```
+
+At this stage, if you log onto Control Center, you should see a running connector:
+
+![connector](./connect_image.png)
+
+### Update CDC enabled 'Person' table
+Now we will send a SQL Command that will update all users in the person.Person table on the AdventureWorks database:
+
+```shell
+./update_person.sh
+Context "minikube" modified.
+Changed database context to 'AdventureWorks'.
+
+(19972 rows affected)
+```
+If you observe the automatically created topic `adventureworks-connect.Person.Person` you will see the update event messages streaming through
+
+
+![topic_update](./topic_update.png)

examples/custom-connect-sql/connect_image.png

@@ -1,4 +1,3 @@
 #!/bin/bash
-nohup kubectl port-forward -n destination connect-0 8083:8083 &
-sleep 2
-curl -XPOST -H "Content-Type: application/json" --data @prod-mssql-connector.json https://localhost:8083/connectors -kv
+export CONNECTOR_CONFIG=$(cat ./prod-mssql-connector.json)
+kubectl exec -i connect-0 -c connect -- curl -k -X PUT -H 'Content-Type:application/json' -d "$CONNECTOR_CONFIG" https://localhost:8083/connectors/debezium-sql-server/config

examples/custom-connect-sql/deploy_connector.sh

@@ -1,9 +1,6 @@
 #!/bin/bash
-
-kubectl create ns sandbox
 kubectl config set-context --current --namespace=sandbox
-
-kubectl exec -it sql-server -- \
+kubectl exec -i sql-server -- \
 tee -a /tmp/person.sql > /dev/null <<EOT
 USE AdventureWorks;
 GO

examples/custom-connect-sql/cdc.sh renamed to examples/custom-connect-sql/enable_cdc.sh

@@ -1,6 +1,4 @@
 {
-  "name": "prod-mssql-connector",
-  "config": {
     "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector", 
 
     "database.hostname": "sql-server.sandbox.svc.cluster.local", 
@@ -40,8 +38,8 @@
 	"time.precision.mode": "connect",
 	"database.history.skip.unparseable.ddl": false,
 	"database.history.store.only.monitored.tables.ddl": false,
-	"table.include.list": ["person.person"],
+	"table.include.list": "person.Person",
 	"table.ignore.builtin": false,
 	"include.schema.changes": true
-  }
 }
+

examples/custom-connect-sql/img.png

@@ -0,0 +1,21 @@
+{
+    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
+    "connection.url": "jdbc:sqlserver://DB_HOST;databaseName=DB_NAME",
+    "connection.user": "DATABASE_USER",
+    "connection.password": "DATABASE_PASS",
+    "db.timezone": "Asia/Riyadh",
+    "mode": "incrementing",
+    "incrementing.column.name": "TRANS_ID",
+    "query": "SELECT * FROM T_PRT_TRANSACTION_V",
+    "topic.prefix": "transaction",
+
+    "principal.service.name": "CONNECT_USER",
+    "principal.service.password": "CONNECT_PASS",  
+
+    "transforms": "createKey,extractInt",
+    "transforms.createKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
+    "transforms.createKey.fields": "JOURNAL_NUMBER",
+    "transforms.extractInt.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
+    "transforms.extractInt.field": "JOURNAL_NUMBER"
+
+  }

examples/custom-connect-sql/prod-mssql-connector.json

@@ -0,0 +1,6 @@
+use AdventureWorks
+GO
+update person.Person
+set ModifiedDate = GETDATE()
+WHERE 1=1
+GO

examples/custom-connect-sql/topic_update.png

@@ -0,0 +1,4 @@
+#!/bin/bash
+kubectl config set-context --current --namespace=sandbox
+export UPDATE_CONFIG=$(cat ./update.sql)
+kubectl exec sql-server -n sandbox -- /opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "nbBg8G4DkR83Xs" -q "$UPDATE_CONFIG" && exit