Streaming data from SQL Server to Kafka to Snowflake ❄️ with Kafka Connect

USE [demo]
GO
EXEC sys.sp_cdc_enable_db
GO 

USE [demo]

EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',
@source_name   = N'ORDERS',
@role_name     = NULL,
@supports_net_changes = 0
GO 

confluent-hub install --no-prompt debezium/debezium-connector-sqlserver:0.10.0

$ curl -s localhost:8083/connector-plugins|jq '.[].class'|grep debezium
"io.debezium.connector.sqlserver.SqlServerConnector"

$ ccloud kafka topic create --partitions 1 dbz_dbhistory.mssql.asgard-01
$ ccloud kafka topic create mssql-01-mssql.dbo.ORDERS
$ ccloud kafka topic list
                 Name
+-------------------------------------+
  dbz_dbhistory.mssql.asgard-01
  mssql-01-mssql.dbo.ORDERS

curl -i -X PUT -H  "Content-Type:application/json" \
    http://localhost:8083/connectors/source-debezium-mssql-01/config \
    -d '{
    "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector", 
    "database.hostname": "mssql",
    "database.port": "1433",
    "database.user": "sa",
    "database.password": "Admin123",
    "database.dbname": "demo",
    "database.server.name": "mssql",
    "table.whitelist":"dbo.orders",
    "database.history.kafka.bootstrap.servers": "${file:/data/credentials.properties:CCLOUD_BROKER_HOST}:9092",
    "database.history.kafka.topic": "dbz_dbhistory.mssql.asgard-01",
    "database.history.consumer.security.protocol": "SASL_SSL",
    "database.history.consumer.ssl.endpoint.identification.algorithm": "https",
    "database.history.consumer.sasl.mechanism": "PLAIN",
    "database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"${file:/data/credentials.properties:CCLOUD_API_KEY}\" password=\"${file:/data/credentials.properties:CCLOUD_API_SECRET}\";",
    "database.history.producer.security.protocol": "SASL_SSL",
    "database.history.producer.ssl.endpoint.identification.algorithm": "https",
    "database.history.producer.sasl.mechanism": "PLAIN",
    "database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"${file:/data/credentials.properties:CCLOUD_API_KEY}\" password=\"${file:/data/credentials.properties:CCLOUD_API_SECRET}\";",
    "decimal.handling.mode":"double",
    "transforms": "unwrap,addTopicPrefix",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.addTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.addTopicPrefix.regex":"(.*)",
    "transforms.addTopicPrefix.replacement":"mssql-01-$1"
    }'

$ ccloud kafka topic consume --from-beginning mssql-04-mssql.dbo.ORDERS
Starting Kafka Consumer. ^C to exit
����������@Proper Job
���q=
ףp�?Wainwright
��Ҝ333333@Proper Job
�ޜ��Q��@Galena

source .env
docker run --rm edenhill/kafkacat:1.5.0 \
    -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \
    -X ssl.ca.location=./etc/ssl/cert.pem -X api.version.request=true \
    -b ${CCLOUD_BROKER_HOST}:9092 \
    -X sasl.username="${CCLOUD_API_KEY}" \
    -X sasl.password="${CCLOUD_API_SECRET}" \
    -r https://"${CCLOUD_SCHEMA_REGISTRY_API_KEY}":"${CCLOUD_SCHEMA_REGISTRY_API_SECRET}"@${CCLOUD_SCHEMA_REGISTRY_HOST} \
    -s avro \
    -t mssql-04-mssql.dbo.ORDERS \
    -f 'Topic %t[%p], offset: %o, Headers: %h, key: %k, payload: %S bytes: %s\n' \
    -C -o beginning -c5

Topic mssql-04-mssql.dbo.ORDERS[5], offset: 0, Headers: , key: , payload: 34 bytes: {"order_id": {"int": 5}, "customer_id": {"int": 14}, "order_ts": {"int": 18233}, "order_total_usd": {"double": 3.8900000000000001}, "item": {"string": "Wainwright"}}
Topic mssql-04-mssql.dbo.ORDERS[5], offset: 1, Headers: , key: , payload: 30 bytes: {"order_id": {"int": 6}, "customer_id": {"int": 16}, "order_ts": {"int": 18225}, "order_total_usd": {"double": 3.9100000000000001}, "item": {"string": "Galena"}}
Topic mssql-04-mssql.dbo.ORDERS[5], offset: 2, Headers: , key: , payload: 32 bytes: {"order_id": {"int": 7}, "customer_id": {"int": 19}, "order_ts": {"int": 18227}, "order_total_usd": {"double": 4.6900000000000004}, "item": {"string": "Landlord"}}
Topic mssql-04-mssql.dbo.ORDERS[5], offset: 3, Headers: , key: , payload: 34 bytes: {"order_id": {"int": 8}, "customer_id": {"int": 2}, "order_ts": {"int": 18228}, "order_total_usd": {"double": 3.6699999999999999}, "item": {"string": "Proper Job"}}
Topic mssql-04-mssql.dbo.ORDERS[5], offset: 4, Headers: , key: , payload: 39 bytes: {"order_id": {"int": 9}, "customer_id": {"int": 5}, "order_ts": {"int": 18229}, "order_total_usd": {"double": 2.2400000000000002}, "item": {"string": "Black Sheep Ale"}}