Debezium PostgreSQL Connector 的快照模式 snapshot.mode

Debezium 的 PostgreSQL 连接器在第一次启动时，默认会执行数据库的初始一致性快照，相当于导出全量数据再导入到Kafka。
可以通过设置连接器配置属性 snapshot.mode (默认：initial) 的值来自定义连接器执行快照的方式。

当 snapshot.mode 设置为 (默认：initial) 时，连接器完成以下任务来创建快照：

1. 使用 SERIALIZABLE、READ ONLY、DEFERRABLE 隔离级别启动事务，以确保此事务中的后续读取针对数据的单个一致版本。 由于其他客户端的后续 INSERT、UPDATE 和 DELETE 操作而对数据进行的任何更改对此事务不可见。
2. 读取数据库事务日志中的当前位置(LSN)。
3. 扫描数据库表和模式，为每一行生成一个 READ 事件并将该事件写入适当的特定于表的 Kafka 主题(topic)。
4. 提交事务。
5. 在连接器偏移(offsets)中记录快照的成功完成。

如果连接器出现故障、重新平衡或在 步骤1 开始之后但在 步骤5 完成之前停止，则在重新启动时连接器将开始一个新的快照。 连接器完成其初始快照后，PostgreSQL 连接器会继续从其在 步骤2 中读取的位置进行流式传输。这可确保连接器不会错过任何更新。 如果连接器由于任何原因再次停止，则在重新启动时，连接器会继续从之前停止的位置流式传输更改。

snapshot.mode 支持的参数配置

图示

按自己的理解，总结了一张图，不一定准确

snapshot.mode = initial

• 向 Kafka Connect 注册并启动一个新的 Debezium PostgreSQL Connector，添加选项 “snapshot.mode”: “initial”

[root@docker ~]# cat pgsql-snapshot-mode.json
{
    "name": "snapshot-mode-initial",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname" : "postgres",
        "database.port" : "5432",
        "database.user" : "postgres",
        "database.password" : "postgres",
        "database.dbname" : "postgres",
        "database.server.name" : "initial",
        "slot.name": "initial_slot",
        "schema.include.list": "inventory",
        "snapshot.mode": "initial",
        "plugin.name": "pgoutput",
        "publication.name": "dbz_inventory_connector",
        "publication.autocreate.mode": "filtered"
    }
}

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @pgsql-snapshot-mode.json

• 源端数据库里的数据情况，inventory schema下有6张表，其中一张表 spatial_ref_sys 不同步
  

• Debezium PostgreSQL Connector 第一次启动后执行初始一致性快照，将全量数据导出转换写入到Kafka，可以看到 Kafka 中为每张有数据的表建了一个 Topics，Topics 的名称格式是 <serverName.schemaName.tableName>，其中表 spatial_ref_sys 不同步，还不知道啥原因，表里的每条数据在 Kafka 中存储为一条消息(Messages)，以下截图也可以看出每个 Topics 的 Messages 个数与 PostgreSQL 中表的数据行数一致。
  

• 源端数据库执行 DML 操作，自动同步到 Kafka

postgres=# insert into inventory.orders values (11001,now(),1003,1,102);

postgres=# update inventory.orders set quantity=2 where id=11001;

postgres=# delete from inventory.orders where id = 11001;

snapshot.mode = always

• 更新上面创建的连接器 snapshot-mode-initial，更改选项 “snapshot.mode”: “always”，每次启动都会执行初始化快照，快照完成后会继续捕获变更数据

[root@docker ~]# cat pgsql-snapshot-mode-update.json
{
    "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname" : "postgres",
    "database.port" : "5432",
    "database.user" : "postgres",
    "database.password" : "postgres",
    "database.dbname" : "postgres",
    "database.server.name" : "initial",
    "slot.name": "initial_slot",
    "schema.include.list": "inventory",
    "snapshot.mode": "always",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_inventory_connector",
    "publication.autocreate.mode": "filtered"
}

curl -i -X PUT -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/snapshot-mode-initial/config -d @pgsql-snapshot-mode-update.json

# 需要重新启动连接器实例和任务实例，只重启连接器实例是不起作用的
curl -s localhost:8083/connectors/snapshot-mode-initial/status | jq
curl -s -X POST localhost:8083/connectors/snapshot-mode-initial/restart?includeTasks=true

• 更新前 Kafka 里面的数据情况
  
• 更新后 Kafka 里面的数据情况，很明显新增了初始化的数据
  

snapshot.mode = always 恢复被删除的 Topics

• 删除现有的 Topics

# 先把连接器停了，不然有时删了 Topics 还会自动创建
curl -s -X DELETE localhost:8083/connectors/snapshot-mode-initial
# 删除 Topics
# docker exec -it connect bash
bin/kafka-topics.sh --bootstrap-server kafka:9092 --list
bin/kafka-topics.sh --bootstrap-server kafka:9092 --delete --topic initial.inventory.customers
bin/kafka-topics.sh --bootstrap-server kafka:9092 --delete --topic initial.inventory.geom
bin/kafka-topics.sh --bootstrap-server kafka:9092 --delete --topic initial.inventory.orders
bin/kafka-topics.sh --bootstrap-server kafka:9092 --delete --topic initial.inventory.products
bin/kafka-topics.sh --bootstrap-server kafka:9092 --delete --topic initial.inventory.products_on_hand

• 重洗启动连接器 snapshot-mode-initial，包含选项 “snapshot.mode”: “always”

[root@docker ~]# cat pgsql-snapshot-mode.json
{
    "name": "snapshot-mode-initial",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname" : "postgres",
        "database.port" : "5432",
        "database.user" : "postgres",
        "database.password" : "postgres",
        "database.dbname" : "postgres",
        "database.server.name" : "initial",
        "slot.name": "initial_slot",
        "schema.include.list": "inventory",
        "snapshot.mode": "always",
        "plugin.name": "pgoutput",
        "publication.name": "dbz_inventory_connector",
        "publication.autocreate.mode": "filtered"
    }
}

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @pgsql-snapshot-mode.json

• 源端数据库执行 DML 操作，同步正常，忽略测试过程
• 需要注意，一切正常了以后，要考虑是否将 snapshot.mode 改成默认的 initial ，否则每次启动连接器都会执行初始化任务。

[root@docker ~]# cat pgsql-snapshot-mode-update.json
{
    "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname" : "postgres",
    "database.port" : "5432",
    "database.user" : "postgres",
    "database.password" : "postgres",
    "database.dbname" : "postgres",
    "database.server.name" : "initial",
    "slot.name": "initial_slot",
    "schema.include.list": "inventory",
    "snapshot.mode": "initial",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_inventory_connector",
    "publication.autocreate.mode": "filtered"
}

curl -i -X PUT -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/snapshot-mode-initial/config -d @pgsql-snapshot-mode-update.json
curl -s -X POST localhost:8083/connectors/snapshot-mode-initial/restart?includeTasks=true
# 已测试，更改 "snapshot.mode": "initial"，重启连接器和任务后数据不会重新初始化，DML 捕获正常。

snapshot.mode = initial_only

• 向 Kafka Connect 注册并启动一个新的 Debezium PostgreSQL Connector，添加选项 “snapshot.mode”: “initial_only”

[root@docker ~]# cat pgsql-snapshot-mode.json
{
    "name": "snapshot-mode-initial_only",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname" : "postgres",
        "database.port" : "5432",
        "database.user" : "postgres",
        "database.password" : "postgres",
        "database.dbname" : "postgres",
        "database.server.name" : "initial_only",
        "slot.name": "initial_only_slot",
        "schema.include.list": "inventory",
        "snapshot.mode": "initial_only",
        "plugin.name": "pgoutput",
        "publication.name": "dbz_inventory_connector",
        "publication.autocreate.mode": "filtered"
    }
}

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @pgsql-snapshot-mode.json

• 源端数据库里的数据情况，inventory schema下有6张表，其中一张表 spatial_ref_sys 不同步
  
• Debezium Oracle Connector 第一次启动后执行初始一致性快照
  
• 源端数据库执行 DML 操作，此时不会同步一致性快照以后的变更数据

postgres=# insert into inventory.orders values (11001,now(),1003,1,102);
postgres=# update inventory.orders set quantity=2 where id=11001;
postgres=# delete from inventory.orders where id = 11001;

• 可以看到 snapshot.mode = initial 的表已经同步数据了，但是 snapshot.mode = initial_only 的表并没有同步数据
  

snapshot.mode = never

• 向 Kafka Connect 注册并启动一个新的 Debezium PostgreSQL Connector，添加选项 “snapshot.mode”: “never”

[root@docker ~]# cat pgsql-snapshot-mode.json
{
    "name": "snapshot-mode-never",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname" : "postgres",
        "database.port" : "5432",
        "database.user" : "postgres",
        "database.password" : "postgres",
        "database.dbname" : "postgres",
        "database.server.name" : "never",
        "slot.name": "never_slot",
        "schema.include.list": "inventory",
        "snapshot.mode": "never",
        "plugin.name": "pgoutput",
        "publication.name": "dbz_inventory_connector",
        "publication.autocreate.mode": "filtered"
    }
}

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @pgsql-snapshot-mode.json

• Debezium PostgreSQL Connector 第一次启动后不会执行初始一致性快照，可以看到 Kafka 的没有存放数据的 Topics
  
• 但是源端数据库执行 DML 操作，Debezium PostgreSQL Connector 会自动捕获变更数据在 Kafka 上创建 Topics 并写入消息

postgres=# insert into inventory.orders values (11001,now(),1003,1,102);
postgres=# update inventory.orders set quantity=2 where id=11001;
postgres=# delete from inventory.orders where id = 11001;

总结一点

Debezium PostgreSQL Connector 比 Debezium Oracle Connector 要好用多了，同步延迟很小，测试还算比较顺利，可以看出 Debezium 对 PostgreSQL 兼容要比 Oracle 要好很多。