Debezium mssql source

​

Quick start

​

Prerequisites

1. Create a MSSQL service: This connector uses the debezium v1.9, Please refer to this document to see the compatible SQL Server versions.
2. Prepare SQL Server: Please refer to this document to complete the prepare steps.
3. Enable CDC for SQL Server: Please refer to this document
4. Enable SQL Server Agent: Please refer to this document
5. Configure topic retention policies: Before running the connector, you must ensure that you have set an infinite retention policy for both the offset.storage.topic and database.history.pulsar.topic. Refer to the Used Topic On Pulsar section for more details.

​

1. Prepare table and permission on SQL Server

CREATE DATABASE mydb;
CREATE TABLE MyTable (id INT PRIMARY KEY, name NVARCHAR(50));
GRANT SELECT ON dbo.MyTable TO {{Your hostname of SQL Server}};

​

2. Create a connector

pulsarctl sources create \
  --source-type debezium-mssql \
  --name debezium-mssql \
  --tenant public \
  --namespace default \
  --parallelism 1 \
  --source-config \
  '{
    "database.hostname": "Your hostname of SQL Server",
    "database.port": "Your port of SQL Server",
    "database.user": "Your user of SQL Server",
    "database.password": "Your password of SQL Server",
    "database.dbname": "Your dbname of SQL Server",
    "table.whitelist": "public.io-test",
    "database.server.name": "mydbserver"
  }'

​

3. Insert and update a data to table

USE mydb;
INSERT INTO dbo.testtable
           ([id]
           ,[name])
     VALUES (1, 'Zike Yang');
GO

​

4. Show data using Pulsar client

bin/pulsar-client \
--url "Your Pulsar serviceUrl" \
consume "persistent://public/default/mssql-test.dbo.testtable" -s "test-sub" -n 0 -p Earliest

----- got message -----
key:[eyJpZCI6MTB9], properties:[], content:{"before":null,"after":{"id":1,"name":"Zike Yang"},"source":{"version":"1.9.7.Final","connector":"sqlserver","name":"mssql-test","ts_ms":1701424975073,"snapshot":"false","db":"mydb","sequence":null,"schema":"dbo","table":"testtable","change_lsn":"00000027:000005a0:0002","commit_lsn":"00000027:000005a0:0003","event_serial_no":1},"op":"c","ts_ms":1701424977325,"transaction":null}

​

Configuration Properties

​

Advanced features

​

Converter options

• org.apache.kafka.connect.json.JsonConverter Thejson-with-envelope config is valid only for the JsonConverter. By default, the value is set to false. When the json-with-envelope value is set to false, the consumer uses the schema Schema.KeyValue(Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED), and the message only consists of the payload. When the json-with-envelope value is set to true, the consumer uses the schema Schema.KeyValue(Schema.BYTES, Schema.BYTES), and the message consists of the schema and the payload.
• org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter If you select the AvroConverter, the consumer uses the schema Schema.KeyValue(Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED), and the message consists of the payload.

​

Used topic on Pulsar

• One topic for storing the database metadata messages. It is named with the database server name ( database.server.name), like public/default/database.server.name.
• One topic (offset.storage.topic) for storing the offset metadata messages. The connector saves the last successfully-committed offsets on this topic.
• One topic (database.history.pulsar.topic) for storing the database history information. The connector writes and recovers DDL statements on this topic.
• One per-table topic. You can set multiple tables for “table.whitelist”, and the connector will send data from each table to a different topic of pulsar. The topic naming rule is: {{database.server.name}}.{{table.name}}. For examples: public/default/mydbserver.public.io-test.

• database.history.pulsar.topic: "{tenant}/{namespace}/{connector-name}-debezium-history-topic"
• offset.storage.topic: "{tenant}/{namespace}/{connector-name}-offset-storage-topic"

pulsar-admin topicPolicies set-retention -s -1 -t -1 ${topic_name}