Merge pull request #9 from AlessandroLollo/update-demo-for-debezium-2.7

Updated code and README to run with Debezium 2.7

Author: yorek

.gitignore

@@ -332,4 +332,7 @@ ASALocalRun/
 # Additional files
 **.env
 __blobstorage__/
-__azurite_db_*
+__azurite_db_*
+
+# SQL Server Debezium connector config (generated)
+sqlserver-connector-config.json

README.md

@@ -70,17 +70,46 @@ The script `01-enable-cdc.sql` enable Change Data Capture on the aforementioned
 All data gathered by Change Data Capture will be send to Event Hubs, so create an Azure Event Hubs in your Azure Subscription. Using the [Azure Cloud Shell](https://shell.azure.com/) Bash:
 
 ```bash
-# create group
-az group create -n debezium -l eastus
+# Set vars
+RESOURCE_GROUP=debezium
+LOCATION=eastus
+EVENTHUB_NAMESPACE=debezium
+EVENTHUB_SCHEMA_HISTORY=schemahistory
 
-# create eventhuvbs with kafka enabled
-az eventhubs namespace create -n debezium -g debezium -l eastus --enable-kafka
+# create group
+az group create \
+    --name $RESOURCE_GROUP \
+    --location $LOCATION
+
+# create eventhub namespace with kafka enabled
+az eventhubs namespace create \
+    --name $EVENTHUB_NAMESPACE \
+    --resource-group $RESOURCE_GROUP \
+    --location $LOCATION \
+    --enable-kafka
+
+# create eventhub for schema history
+az eventhubs eventhub create \
+    --resource-group $RESOURCE_GROUP \
+    --namespace $EVENTHUB_NAMESPACE \
+    --name $EVENTHUB_SCHEMA_HISTORY \
+    --partition-count 1 \
+    --cleanup-policy Delete \
+    --retention-time-in-hours 168
 ```
 
 Later in the configuration process you'll need the EventHubs connection string, so grab it and store it somewhere:
 
 ```bash
-az eventhubs namespace authorization-rule keys list -g debezium --namespace-name debezium -n RootManageSharedAccessKey --query "primaryConnectionString" -o tsv
+RESOURCE_GROUP=debezium
+EVENTHUB_NAMESPACE=debezium
+
+az eventhubs namespace authorization-rule keys list \
+    --resource-group $RESOURCE_GROUP \
+    --namespace-name $EVENTHUB_NAMESPACE \
+    --name RootManageSharedAccessKey \
+    --query "primaryConnectionString" \
+    --output tsv
  ```
 
 ### Run Debezium
@@ -96,12 +125,12 @@ If prefer a more lean and quick easy to start using Debezium, you can just use t
 Docker Compose will use `.env` to get the environment variables values used in the `.yaml` configuration file. The provided `.env.template` file look like the following:
 
 ```bash
-DEBEZIUM_VERSION=1.6
-EH_NAME=debezium
-EH_CONNECTION_STRING=
+DEBEZIUM_VERSION=2.7
+EVENTHUB_NAMESPACE=<eventhub_namespace>
+EVENTHUB_CONNECTION_STRING=<eventhub_connection_string>
 ```
 
-Copy it and create a new `.env` file. Leave the version set to 1.6. Change the `EH_NAME` to the EventHubs name you created before. Also set `EH_CONNECTION_STRING` to hold the EventHubs connection string you got before. Make sure not to use any additional quotes or double quotes.
+Copy it and create a new `.env` file. Leave the version set to `2.7`. Change the `EVENTHUB_NAMESPACE` to the EventHubs name you created before. Also set `EVENTHUB_CONNECTION_STRING` to hold the EventHubs connection string you got before. Make sure not to use any additional quotes or double quotes.
 
 #### The .yaml file
 
@@ -154,38 +183,54 @@ Once the startup has finished, you'll see something like
 [Worker clientId=connect-1, groupId=1] Finished starting connectors and tasks   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
 ```
 
-you will see three topics (or EventHub to use the Azure EventHubs nomenclature):
+you will see four topics (or EventHub to use the Azure EventHubs nomenclature):
 
 ```bash
-az eventhubs eventhub list -g debezium --namespace debezium -o table
+
+RESOURCE_GROUP=debezium
+EVENTHUB_NAMESPACE=debezium
+
+az eventhubs eventhub list \
+    --resource-group $RESOURCE_GROUP \
+    --namespace-name $EVENTHUB_NAMESPACE \
+    --output table
 ```
 
 and the result will show:
 
 - debezium_configs
 - debezium_offsets
 - debezium_statuses
+- schemahistory
 
 to explore Azure Event Hubs is strongly suggest to download and use [Service Bus Explorer](https://github.com/paolosalvatori/ServiceBusExplorer)
 
 #### Register SQL Server Connector
 
-Now that Debezium is running, the SQL Server Connector (which is used both for connecting to Azure SQL or SQL Server) can be registered. Before doing that, make sure to specify the correct connection for your SQL Server instance in the `debezium/register-sqlserver-eh.json` file. You can create one using the provided `.template` file.
+Now that Debezium is running, the SQL Server Connector (which is used both for connecting to Azure SQL or SQL Server) can be registered. Before doing that, make sure to specify the correct connection for your SQL Server instance in a file named `sqlserver-connector-config.json`. You can create one using the template file [sqlserver-connector-config.json.template](debezium/sqlserver-connector-config.json.template) file.
 
-If you are using the Wide World Importers database, the only values you have to change are:
+Make sure to change the following properties to match your SQL Server configuration:
 
 ```json
-"database.hostname": "<server>.database.windows.net",
-"database.dbname": "<database-name>",
+"database.hostname": "<sql_server_name>.database.windows.net",
+"database.names": "<db_name>",
+"database.user" : "<debezium_user_name>",
+"database.password" : "<debezium_user_password>",
 ```
 
-If you are following the step-by-step guide using a database of yours, make sure to also correctly set values for
+You would need to replace `<sql_server_name>`, `<db_name>`, `<debezium_user_name>`, and `<debezium_user_password>` with proper values.
+
+Also, make sure to change the following properties to match your Azure Event Hub configuration:
 
 ```json
-"database.user" : "debezium-wwi",
-"database.password" : "Abcd1234!",
+"schema.history.internal.kafka.bootstrap.servers": "<eventhub_namespace>.servicebus.windows.net:9093",
+"schema.history.internal.kafka.topic": "<eventhub_schema_history>",
+"schema.history.internal.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"<eventhub_connectionstring>\";",
+"schema.history.internal.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"<eventhub_connectionstring>\";",
 ```
 
+You would need to replace `<eventhub_namespace>`, `<eventhub_schema_history>`, and `<eventhub_connectionstring>` with proper values.
+
 All the other values used are explained in detail here:
 
 [SQL Server Connector Configuration Values](./documentation/SQL-Server-Connector-Configuration-Value.md)
@@ -253,6 +298,9 @@ Congratulations, you now have a working Change Stream from SQL Server. This open
 
 If you're using Debezium with Azure SQL MI or Azure SQL DB, you may want to run Debezium on Azure. Sample script to run the Debezium container on Azure Container Instances are available in the `debezium/azure` folder.
 
+Please note that the script [01-register-connector.sh](debezium/azure/01-register-connector.sh) will take care of creating the Debezium SQL Server connector configuration JSON file, and then will use it to register a new connector on Debezium.
+The generated file will be stored in the `debezium` folder.
+
 ### Connector Configuration
 
 More details on SQL Server and Event Hubs specific configuration here:

debezium/azure/00-deploy-debezium.sh

@@ -3,28 +3,51 @@
 # Strict mode, fail on any error
 set -euo pipefail
 
-export DEBEZIUM_VERSION=1.8
-export RESOURCE_GROUP="dm-debezium"
-export EVENTHUB_NAME="dm-debezium"
-export CONTAINER_NAME="dm-debezium"
+DEBEZIUM_VERSION=2.7
+RESOURCE_GROUP="debezium"
+EVENTHUB_NAMESPACE="debezium"
+EVENTHUB_SCHEMA_HISTORY="schemahistory"
+CONTAINER_NAME="debezium"
+LOCATION="WestUS2"
 
 echo "deploying resource group"
-az group create -n $RESOURCE_GROUP -l WestUS2
+az group create \
+	--name $RESOURCE_GROUP \
+	--location $LOCATION
 
 echo "deploying eventhubs namespace"
-az eventhubs namespace create -g $RESOURCE_GROUP -n $EVENTHUB_NAME --enable-kafka=true -l WestUS2
+az eventhubs namespace create \
+	--resource-group $RESOURCE_GROUP \
+	--location $LOCATION \
+	--name $EVENTHUB_NAMESPACE \
+	--enable-kafka=true
 
-echo "gathering eventhubs info"
-export EH_NAME=`az eventhubs namespace list -g $EVENTHUB_NAME --query '[].name' -o tsv`
-export EH_CONNECTION_STRING=`az eventhubs namespace authorization-rule keys list -g $CONTAINER_NAME -n RootManageSharedAccessKey --namespace-name $EVENTHUB_NAME -o tsv --query 'primaryConnectionString'`
+echo "deploying schema history event hub"
+az eventhubs eventhub create \
+    --resource-group $RESOURCE_GROUP \
+    --namespace-name $EVENTHUB_NAMESPACE \
+    --name $EVENTHUB_SCHEMA_HISTORY \
+    --partition-count 1 \
+    --cleanup-policy Delete \
+    --retention-time-in-hours 168 \
+    --output none
+
+echo "gathering eventhubs connection string"
+EVENTHUB_CONNECTION_STRING=`az eventhubs namespace authorization-rule keys list --resource-group $RESOURCE_GROUP --name RootManageSharedAccessKey --namespace-name $EVENTHUB_NAMESPACE --output tsv --query 'primaryConnectionString'`
 
 echo "deploying debezium container"
-az container create -g $RESOURCE_GROUP -n $CONTAINER_NAME \
+az container create \
+	--resource-group $RESOURCE_GROUP \
+	--location $LOCATION \
+	--name $CONTAINER_NAME \
 	--image debezium/connect:${DEBEZIUM_VERSION} \
-	--ports 8083 --ip-address Public \
-	--os-type Linux --cpu 2 --memory 4 \
+	--ports 8083 \
+	--ip-address Public \
+	--os-type Linux \
+	--cpu 2 \
+	--memory 4 \
 	--environment-variables \
-		BOOTSTRAP_SERVERS=${EH_NAME}.servicebus.windows.net:9093 \
+		BOOTSTRAP_SERVERS=${EVENTHUB_NAMESPACE}.servicebus.windows.net:9093 \
 		GROUP_ID=1 \
 		CONFIG_STORAGE_TOPIC=debezium_configs \
 		OFFSET_STORAGE_TOPIC=debezium_offsets \
@@ -34,13 +57,13 @@ az container create -g $RESOURCE_GROUP -n $CONTAINER_NAME \
 		CONNECT_REQUEST_TIMEOUT_MS=60000 \
 		CONNECT_SECURITY_PROTOCOL=SASL_SSL \
 		CONNECT_SASL_MECHANISM=PLAIN \
-		CONNECT_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\$ConnectionString\" password=\"${EH_CONNECTION_STRING}\";" \
+		CONNECT_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\$ConnectionString\" password=\"${EVENTHUB_CONNECTION_STRING}\";" \
 		CONNECT_PRODUCER_SECURITY_PROTOCOL=SASL_SSL \
 		CONNECT_PRODUCER_SASL_MECHANISM=PLAIN \
-		CONNECT_PRODUCER_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\$ConnectionString\" password=\"${EH_CONNECTION_STRING}\";" \
+		CONNECT_PRODUCER_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\$ConnectionString\" password=\"${EVENTHUB_CONNECTION_STRING}\";" \
 		CONNECT_CONSUMER_SECURITY_PROTOCOL=SASL_SSL \
 		CONNECT_CONSUMER_SASL_MECHANISM=PLAIN \
-		CONNECT_CONSUMER_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\$ConnectionString\" password=\"${EH_CONNECTION_STRING}\";"
+		CONNECT_CONSUMER_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\$ConnectionString\" password=\"${EVENTHUB_CONNECTION_STRING}\";"
 
 echo "eventhub connection string"
-echo $EH_CONNECTION_STRING
+echo $EVENTHUB_CONNECTION_STRING

debezium/azure/01-register-connector.sh

@@ -3,14 +3,14 @@
 # Strict mode, fail on any error
 set -euo pipefail
 
-export RESOURCE_GROUP="dm-debezium"
-export CONTAINER_NAME="dm-debezium"
+RESOURCE_GROUP="debezium"
+CONTAINER_NAME="debezium"
 
 echo "finding debezium ip"
-export DEBEZIUM_IP=`az container show -g $RESOURCE_GROUP -n $CONTAINER_NAME -o tsv --query "ipAddress.ip"`
+DEBEZIUM_IP=`az container show --resource-group $RESOURCE_GROUP --name $CONTAINER_NAME --output tsv --query "ipAddress.ip"`
 
 echo "registering connector"
 curl -i -X POST \
     -H "Accept:application/json" -H  "Content-Type:application/json" \
     http://${DEBEZIUM_IP}:8083/connectors/ \
-    -d @../register-sqlserver-eh.json
+    -d @../sqlserver-connector-config.json

debezium/on-prem/.env.sample

@@ -1,3 +1,3 @@
-DEBEZIUM_VERSION=1.6
-EH_NAME=debezium
-EH_CONNECTION_STRING=Endpoint=sb://.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=
+DEBEZIUM_VERSION=2.7
+EVENTHUB_NAMESPACE=<eventhub_namespace>
+EVENTHUB_CONNECTION_STRING=<eventhub_connection_string>

debezium/on-prem/.env.template

@@ -5,7 +5,7 @@ services:
     ports:
      - 8083:8083
     environment:
-     - BOOTSTRAP_SERVERS=${EH_NAME}.servicebus.windows.net:9093
+     - BOOTSTRAP_SERVERS=${EVENTHUB_NAMESPACE}.servicebus.windows.net:9093
      - GROUP_ID=1
      - CONFIG_STORAGE_TOPIC=debezium_configs
      - OFFSET_STORAGE_TOPIC=debezium_offsets
@@ -15,10 +15,10 @@ services:
      - CONNECT_REQUEST_TIMEOUT_MS=60000
      - CONNECT_SECURITY_PROTOCOL=SASL_SSL
      - CONNECT_SASL_MECHANISM=PLAIN
-     - CONNECT_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$$ConnectionString" password="${EH_CONNECTION_STRING}";
+     - CONNECT_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$$ConnectionString" password="${EVENTHUB_CONNECTION_STRING}";
      - CONNECT_PRODUCER_SECURITY_PROTOCOL=SASL_SSL
      - CONNECT_PRODUCER_SASL_MECHANISM=PLAIN
-     - CONNECT_PRODUCER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$$ConnectionString" password="${EH_CONNECTION_STRING}";
+     - CONNECT_PRODUCER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$$ConnectionString" password="${EVENTHUB_CONNECTION_STRING}";
      - CONNECT_CONSUMER_SECURITY_PROTOCOL=SASL_SSL
      - CONNECT_CONSUMER_SASL_MECHANISM=PLAIN
-     - CONNECT_CONSUMER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$$ConnectionString" password="${EH_CONNECTION_STRING}";
+     - CONNECT_CONSUMER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$$ConnectionString" password="${EVENTHUB_CONNECTION_STRING}";

debezium/on-prem/delete-connector.ps1

@@ -1,2 +1,2 @@
-$JSON = Get-Content '..\register-sqlserver-eh.json' | Out-String
-Invoke-RestMethod http://localhost:8083/connectors/ -Method POST -Body $JSON -ContentType "application/json"     
+$JSON = Get-Content '..\sqlserver-connector-config.json' | Out-String
+Invoke-RestMethod http://localhost:8083/connectors/ -Method POST -Body $JSON -ContentType "application/json"

debezium/on-prem/delete-connector.sh

@@ -1,3 +1,8 @@
 #!/bin/sh
 
-curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @../register-sqlserver-eh.json -w "\n"
+curl -i -X POST \
+    -H "Accept:application/json" \
+    -H "Content-Type:application/json" \
+    http://localhost:8083/connectors/ \
+    -d @../sqlserver-connector-config.json \
+    -w "\n"

debezium/on-prem/docker-compose.yaml

@@ -0,0 +1,30 @@
+{
+    "name": "wwi",
+    "config": {
+      "snapshot.mode": "schema_only",
+      "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
+      "database.hostname": "<sql_server_name>.database.windows.net",
+      "database.port": "1433",
+      "database.user": "<debezium_user_name>",
+      "database.password": "<debezium_user_password>",
+      "database.names": "<db_name>",
+      "driver.encrypt": "false",
+      "driver.trustServerCertificate": "true",
+      "schema.history.internal.kafka.bootstrap.servers": "<eventhub_namespace>.servicebus.windows.net:9093",
+      "schema.history.internal.kafka.topic": "<eventhub_schema_history>",
+      "schema.history.internal.consumer.security.protocol": "SASL_SSL",
+      "schema.history.internal.consumer.sasl.mechanism": "PLAIN",
+      "schema.history.internal.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"<eventhub_connectionstring>\";",
+      "schema.history.internal.producer.security.protocol": "SASL_SSL",
+      "schema.history.internal.producer.sasl.mechanism": "PLAIN",
+      "schema.history.internal.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" password=\"<eventhub_connectionstring>\";",
+      "table.include.list": "Sales.Orders,Warehouse.StockItems",
+      "tombstones.on.delete": "false",
+      "topic.prefix": "SQLAzure",
+      "transforms": "Reroute",
+      "transforms.Reroute.type": "io.debezium.transforms.ByLogicalTableRouter",
+      "transforms.Reroute.topic.regex": "(.*)",
+      "transforms.Reroute.topic.replacement": "wwi"
+    }
+  }
+