3.2. Debezium Db2 连接器如何工作

Db2 的复制功能不是存储数据库更改的完整历史记录。因此，当 Debezium Db2 连接器第一次连接到数据库时，它会生成处于捕获模式的表快照，并将这个状态流传输到 Kafka。这会为表内容建立基准。

默认情况下，当 Db2 连接器执行快照时，它会执行以下操作：

SELECT * FROM <tableName> ....

SELECT * FROM <tableName> WHERE <additional-condition> ....

INSERT INTO <signalTable> (id, type, data) VALUES ('<id>', '<snapshotType>', '{"data-collections": ["<tableName>","<tableName>"],"type":"<snapshotType>","additional-condition":"<additional-condition>"}');

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-condition":"color=blue"}');

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-condition":"color=blue AND quantity>10"}');

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 1
    },
    "op":"r", 2
    "ts_ms":"1620393591654",
    "transaction":null
}

完成快照后，当 Debezium Db2 连接器首次启动后，连接器会标识处于捕获模式的每个源表的 change-data 表。连接器为每个 change-data 表执行以下操作：

重启后，连接器会从关闭的偏移（提交并更改 LSN）中恢复发出更改事件。当连接器正在运行并发出更改事件时，如果您从捕获模式中删除表，或者添加表来捕获模式，连接器会检测到更改并相应地修改其行为。

默认情况下，Db2 连接器将表中发生的所有 INSERT、UPDATE 和 DELETE 操作的更改事件写入特定于该表的单个 Apache Kafka 主题。连接器使用以下惯例命名更改事件主题：

topicPrefix.schemaName.tableName

以下列表提供默认名称组件的定义：

例如，一个使用 mydatabase 数据库的 Db2 安装，其中包含四个表：PRODUCTS, PRODUCTS_ON_HAND, CUSTOMERS, 和 ORDERS，它们包括在 MYSCHEMA schema 中。连接器会将事件发送到这四个 Kafka 主题：

连接器应用类似的命名约定来标记其内部数据库架构历史记录主题、架构更改主题 以及 事务元数据主题。

如果默认主题名称不满足您的要求，您可以配置自定义主题名称。要配置自定义主题名称，您可以在逻辑主题路由 SMT 中指定正则表达式。有关使用逻辑主题路由 SMT 自定义主题命名的更多信息，请参阅 主题路由。

您可以配置 Debezium Db2 连接器来生成 schema 更改事件，该事件描述了应用于数据库中捕获的表的 schema 更改。

Debezium 在以下情况下向 schema 更改主题发送信息：

连接器将模式更改事件改为 Kafka 模式更改主题，其名称为 < topicPrefix& gt;，其中 &lt ;topicPrefix > 是 topic.prefix connector 配置属性中指定的主题前缀。连接器发送到 schema 更改主题的消息包含以下元素的有效负载：

{
  "schema": {
  ...
  },
  "payload": {
    "source": {
      "version": "2.1.4.Final",
      "connector": "db2",
      "name": "db2",
      "ts_ms": 0,
      "snapshot": "true",
      "db": "testdb",
      "schema": "DB2INST1",
      "table": "CUSTOMERS",
      "change_lsn": null,
      "commit_lsn": "00000025:00000d98:00a2",
      "event_serial_no": null
    },
    "ts_ms": 1588252618953, 1
    "databaseName": "TESTDB", 2
    "schemaName": "DB2INST1",
    "ddl": null, 3
    "tableChanges": [ 4
      {
        "type": "CREATE", 5
        "id": "\"DB2INST1\".\"CUSTOMERS\"", 6
        "table": { 7
          "defaultCharsetName": null,
          "primaryKeyColumnNames": [ 8
            "ID"
          ],
          "columns": [ 9
            {
              "name": "ID",
              "jdbcType": 4,
              "nativeType": null,
              "typeName": "int identity",
              "typeExpression": "int identity",
              "charsetName": null,
              "length": 10,
              "scale": 0,
              "position": 1,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "FIRST_NAME",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "varchar",
              "typeExpression": "varchar",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 2,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "LAST_NAME",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "varchar",
              "typeExpression": "varchar",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 3,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "EMAIL",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "varchar",
              "typeExpression": "varchar",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 4,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            }
          ],
          "attributes": [ 10
            {
              "customAttribute": "attributeValue"
            }
          ]
        }
      }
    ]
  }
}

在连接器发送到 schema 更改主题的消息中，message 键是包含架构更改的数据库的名称。在以下示例中，payload 字段包含键：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "databaseName"
      }
    ],
    "optional": false,
    "name": "io.debezium.connector.db2.SchemaChangeKey"
  },
  "payload": {
    "databaseName": "TESTDB"
  }
}

Debezium 可以生成代表事务边界的事件，并增强更改数据事件消息。

Debezium 为每个事务生成 BEGIN 和 END 分隔符的事务边界事件。事务边界事件包含以下字段：

{
  "status": "BEGIN",
  "id": "00000025:00000d08:0025",
  "ts_ms": 1486500577125,
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "00000025:00000d08:0025",
  "ts_ms": 1486500577691,
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "testDB.dbo.tablea",
      "event_count": 1
    },
    {
      "data_collection": "testDB.dbo.tableb",
      "event_count": 1
    }
  ]
}

除非通过 topic.transaction 选项覆盖，否则连接器会将事务事件发送到 < topic.prefix>.transaction 主题。

以下是消息的示例：

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "00000025:00000d08:0025",
    "total_order": "1",
    "data_collection_order": "1"
  }
}