解析SQL Server CDC配合Kafka Connect监听数据变化的问题

select * from sys.databases where is_cdc_enabled = 1

create  table T_LioCDC
(
    ID int identity(1,1) primary key ,
    Name nvarchar(16),
    Sex bit,
    CreateTime datetime,
    UpdateTime datetime
);

exec sp_cdc_enable_table 
@source_schema='dbo', 
@source_name='T_LioCDC', 
@role_name=null,
@supports_net_changes = 1;

//kafka集群ip+portbootstrap.servers=172.192.10.210:9092,172.192.10.211:9092,172.192.10.212:9092

key.converter.schemas.enable=false
value.converter.schemas.enable=false

offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
offset.storage.partitions=3
offset.storage.cleanup.policy=compact

config.storage.topic=connect-configs
config.storage.replication.factor=1

status.storage.topic=connect-status
status.storage.replication.factor=1
status.storage.partitions=3
//刚刚下载连接器解压的路径
plugin.path=/usr/soft/kafka/kafka_2.13_2.7.0/plugins

{
    "name": "sqlserver-cdc-source",
    "config": {
        "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
        "database.server.name" : "JnServer",
        "database.hostname" : "172.192.20.2", --目标数据库的ip
        "database.port" : "1433",  --目标数据库的端口
        "database.user" : "sa",   --目标数据库的账号
        "database.password" : "123456",  --密码
        "database.dbname" : "Dis",  --目标数据库的数据库名称
        "table.whitelist": "dbo.T_LioCDC", --监听表名
         "schemas.enable" : "false",  
         "mode":"incrementing",  --增量模式
         "incrementing.column.name": "ID", --增量列名
        "database.history.kafka.bootstrap.servers" : "172.192.10.210:9092,172.192.10.211:9092,172.192.10.212", --kafka集群
        "database.history.kafka.topic": "TopicTLioCDC",  --kafka topic内部使用，不是由消费者使用
        "value.converter.schemas.enable":"false",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter"
    }
}
//源文地址： https://www.cnblogs.com/EminemJK/p/14688907.html

GET /connectors – 返回所有正在运行的connector名。
POST /connectors – 新建一个connector; 请求体必须是json格式并且需要包含name字段和config字段，name是connector的名字，config是json格式，必须包含你的connector的配置信息。
GET /connectors/{name} – 获取指定connetor的信息。
GET /connectors/{name}/config – 获取指定connector的配置信息。
PUT /connectors/{name}/config – 更新指定connector的配置信息。
GET /connectors/{name}/status – 获取指定connector的状态，包括它是否在运行、停止、或者失败，如果发生错误，还会列出错误的具体信息。
GET /connectors/{name}/tasks – 获取指定connector正在运行的task。
GET /connectors/{name}/tasks/{taskid}/status – 获取指定connector的task的状态信息。
PUT /connectors/{name}/pause – 暂停connector和它的task，停止数据处理知道它被恢复。
PUT /connectors/{name}/resume – 恢复一个被暂停的connector。
POST /connectors/{name}/restart – 重启一个connector，尤其是在一个connector运行失败的情况下比较常用
POST /connectors/{name}/tasks/{taskId}/restart – 重启一个task，一般是因为它运行失败才这样做。
DELETE /connectors/{name} – 删除一个connector，停止它的所有task并删除配置。//源文地址： https://www.cnblogs.com/EminemJK/p/14688907.html

--测试代码
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('A',1,getdate(),getdate())
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('B',0,getdate(),getdate())
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('C',1,getdate(),getdate())
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('D',0,getdate(),getdate())
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('E',1,getdate(),getdate())
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('F',1,getdate(),getdate())
insert into T_LioCDC(name, sex, createtime,UpdateTime)  values ('G',0,getdate(),getdate())

update T_LioCDC
set Name='Lio.Huang',UpdateTime=getdate()
where ID=7