5.2. Debezium MongoDB 连接器的工作方式

MongoDB 连接器支持以下 MongoDB 拓扑：

连接器配置属性 topic.prefix 充当 MongoDB 副本集或分片集群的逻辑名称。连接器以多种方式使用逻辑名称： 作为所有主题名称的前缀，并在记录每个副本集的更改流位置时作为唯一标识符。

您应该为每个 MongoDB 连接器分配一个唯一的逻辑名称，以有意义的描述源 MongoDB 系统。我们建议逻辑名称以字母或下划线字符开头，以及字母数字字符或下划线的剩余字符。

当 Debezium 任务开始使用副本集时，它使用连接器的逻辑名称和副本集名称来查找一个 偏移 信息，该偏移之前停止读取更改的位置。如果找到偏移并且仍然存在于 oplog 中，则任务会立即进行 流传输更改，从记录的偏移位置开始。

但是，如果没有找到偏移，或者 oplog 不再包含该位置，则任务必须首先通过执行快照来获取副本集内容的当前状态。这个过程首先记录 oplog 的当前位置，并记录为偏移（以及表示快照已启动的标记）。然后，该任务会继续复制每个集合，尽可能生成多个线程（最多 snapshot.max.threads 配置属性值）来并行执行此功能。连接器为其看到的每个文档记录一个单独的 读取事件。每个读取事件都包含对象的标识符、对象的完整状态和有关找到对象的 MongoDB 副本集 的源 信息。源信息还包括一个标记，表示该事件是在快照期间生成的。

此快照将继续，直到它复制了与连接器的过滤器匹配的所有集合。如果在任务快照完成前停止连接器，重启连接器会在重启连接器再次开始快照。

您可以在以下部分找到有关快照的更多信息：

默认情况下，连接器仅在首次启动后运行初始快照操作。在正常情况下，在这个初始快照后，连接器不会重复快照过程。连接器捕获的任何更改事件数据都只通过流处理。

然而，在某些情况下，连接器在初始快照期间获得的数据可能会过时、丢失或不完整。为了提供重新捕获集合数据的机制，Debezium 包含一个执行临时快照的选项。数据库中的以下更改可能会导致执行临时快照：

您可以通过启动所谓的 临时快照来为之前捕获的集合重新运行快照。临时快照需要使用 信号集合。您可以通过向 Debezium 信号集合发送信号请求来发起临时快照。

当您启动现有集合的临时快照时，连接器会将内容附加到已用于集合的主题。如果删除了之前存在的主题，如果启用了 自动主题创建，Debezium 可以自动创建主题。

临时快照信号指定要包含在快照中的集合。快照可以捕获整个数据库的内容，或者仅捕获数据库中集合的子集。另外，快照也可以捕获数据库中集合内容的子集。

您可以通过将 execute-snapshot 消息发送到信号集合来指定要捕获的集合。将 execute-snapshot 信号类型设置为 增量，并提供快照中包含的集合名称，如下表所述：

目前，execute-snapshot 操作类型仅触发 增量快照。如需更多信息，请参阅 增加快照。

为了提供管理快照的灵活性，Debezium 包含附加快照机制，称为 增量快照。增量快照依赖于 Debezium 机制 向 Debezium 连接器发送信号。

在增量快照中，除了一次捕获数据库的完整状态，就像初始快照一样，Debebe 会在一系列可配置的块中捕获每个集合。您可以指定您希望快照捕获 的集合以及每个块的大小。块大小决定了快照在数据库的每个获取操作期间收集的行数。增量快照的默认块大小为 1024 行。

当增量快照进行时，Debebe 使用 watermarks 跟踪其进度，维护它捕获的每个集合行的记录。与标准初始快照过程相比，捕获数据的阶段方法具有以下优点：

当快照继续进行时，其他进程可能会继续访问数据库，可能会修改集合记录。为了反映此类更改，INSERT、UPDATE 或 DELETE 操作会按照常常提交到事务日志。同样，持续 Debezium 流进程将继续检测这些更改事件，并将相应的更改事件记录发送到 Kafka。

对于每个数据收集，Debezium 会发出两种类型的事件，并将其存储在单个目标 Kafka 主题中。从表直接捕获的快照记录作为 READ 操作发送。同时，当用户继续更新数据收集中的记录，并且会更新事务日志来反映每个提交，Debezium 会为每个更改发出 UPDATE 或 DELETE 操作。

当快照窗口打开时，Debezium 开始处理快照块，它会向内存缓冲区提供快照记录。在快照窗口期间，缓冲区中 READ 事件的主密钥与传入流事件的主键进行比较。如果没有找到匹配项，则流化事件记录将直接发送到 Kafka。如果 Debezium 检测到匹配项，它会丢弃缓冲的 READ 事件，并将流化记录写入目标主题，因为流的事件逻辑地取代静态快照事件。在块关闭的快照窗口后，缓冲区仅包含 READ 事件，这些事件不存在相关的事务日志事件。Debezium 将这些剩余的 READ 事件发送到集合的 Kafka 主题。

连接器为每个快照块重复这个过程。

要将增量快照与分片 MongoDB 集群搭配使用，您必须为以下属性设置特定值：

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

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-condition":"color='blue'"}}`

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-condition":"color='blue' AND brand='MyBrand'"}}`

Key = `test_connector`

Value = `{"type":"stop-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

在副本集的连接器任务记录偏移后，它使用偏移来确定应该开始流更改的 oplog 中的位置。然后，任务（取决于配置）连接到副本集的主节点，或连接到副本集范围更改流，并开始从该位置流更改。它处理所有创建、插入和删除操作，并将其转换为 Debezium 更改事件。每个更改事件都包含找到操作的 oplog 中的位置，连接器会定期将其记录为最新的偏移。记录偏移的时间间隔由 offset.flush.interval.ms 进行管理，它是一个 Kafka Connect worker 配置属性。

当连接器被安全停止时，处理最后一个偏移会被记录，以便在重启后，连接器将继续保持关闭的位置。如果连接器的任务意外终止，则任务可能会在最后一次记录偏移后处理和生成事件，但在记录最后一个偏移前；重启时，连接器从最后 记录的 偏移开始，可能会生成之前在崩溃前生成的一些相同事件。

如前文所述，连接器任务始终使用副本集的主节点从 oplog 中流更改，确保连接器尽可能看到最新的操作，并可以捕获比使用第二个aries 更低的延迟的变化。当副本集选择新主节点时，连接器会立即停止流更改，连接到新主节点，并在同一位置开始从新主节点流更改。同样，如果连接器遇到与副本集成员通信的问题，它会尝试使用 exponential backoff 来重新连接，因此不会大量副本集，并在连接后继续从最后一个离开的位置继续流更改。这样，连接器可以动态地调整副本集成员资格中的更改，并自动处理通信失败。

总之，MongoDB 连接器在大多数情况下继续运行。通信问题可能会导致连接器等待问题解决。

在 MongoDB 6.0 及更高版本中，您可以配置更改流来发出文档的预镜像状态，以填充 MongoDB 更改事件的 before 字段。要在 MongoDB 中启用预镜像，您必须使用 db.createCollection(), create, 或 collMod 为集合设置 changeStreamPreAndPostImages。要启用 Debezium MongoDB 在更改事件中包含预镜像，请将连接器的 capture.mode 设置为其中一个 *_with_pre_image 选项。

MongoDB 连接器将所有插入、更新和删除操作的事件写入每个集合中的文档到单个 Kafka 主题。Kafka 主题的名称始终使用 logicalName.databaseName.collectionName 格式，其中 logicalName 是连接器的逻辑名称（使用 topic.prefix 配置属性指定），databaseName 是创建发生在的数据库的名称，collectionName 是受影响的文档所在的 MongoDB 集合的名称。

例如，假设一个 MongoDB 副本集有一个 inventory 数据库，其中包含四个集合：products, products_on_hand, customers, 和 orders。如果监控这个数据库的连接器有一个逻辑名称 fulfillment，则这个连接器会在这四个 Kafka 主题上生成事件：

请注意，主题名称不包含副本集名称或分片名称。因此，对分片集合（每个分片都包含集合文档的子集）的所有更改都会进入相同的 Kafka 主题。

您可以根据需要将 Kafka 设置为自动创建主题。如果没有，则必须在启动连接器前使用 Kafka 管理工具创建主题。

MongoDB 连接器不会明确确定如何为事件分区主题。相反，它允许 Kafka 如何根据事件密钥确定分区主题。您可以通过在 Kafka Connect worker 配置中定义分区 器 实现的名称来更改 Kafka 的分区逻辑。

Kafka 只为写入单个主题分区的事件维护总顺序。按键对事件进行分区意味着，具有相同键的所有事件始终都到达同一分区。这样可确保特定文档的所有事件始终被完全排序。

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

对于每个事务 BEGIN 和 END，Debezium 会生成一个包含以下字段的事件：

以下示例显示了一个典型的信息：

{
  "status": "BEGIN",
  "id": "1462833718356672513",
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "1462833718356672513",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "rs0.testDB.collectiona",
      "event_count": 1
    },
    {
      "data_collection": "rs0.testDB.collectionb",
      "event_count": 1
    }
  ]
}

除非通过 topic.transaction 选项覆盖，否则事务事件将写入名为 <topic. prefix>.transaction 的主题。

下面是一个信息示例：

{
  "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"annek@noanswer.org\"}",
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "1462833718356672513",
    "total_order": "1",
    "data_collection_order": "1"
  }
}