SQL基础：解析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