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

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

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

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

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

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

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

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

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

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

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

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

在 MongoDB 6.0 及更高版本中，您可以配置更改流，以发出文档的预镜像状态，以填充 MongoDB 更改事件的 before 字段。要在 MongoDB 中启用预镜像，您必须使用 db. create Collection （）、创建或 collMod 为集合设置 changeStreamPreAndPostImages。要使 Debezium MongoDB 在更改事件中包含 pre-images，请将连接器的 capture.mode 设置为 一 一 一 一，with_pre_image 选项。

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

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

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

您可以根据需要将 Kafka 设置为自动创建主题。https://kafka.apache.org/documentation.html#basic_ops_add_topic如果没有，则必须使用 Kafka 管理工具在启动连接器前创建主题。

MongoDB 连接器不会明确决定如何为事件分区主题。相反，它允许 Kafka 根据事件密钥决定如何对主题进行分区。您可以通过在 Kafka Connect worker 配置中定义 Partitioner 实现来更改 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"
  }
}