Debezium MongoDB 连接器

尽管 Debezium MongoDB 连接器不属于副本集，但它使用类似的复制机制来获取 oplog 数据。主要区别在于连接器不直接读取 oplog。相反，它将 oplog 数据的捕获和解码委托给 MongoDB 的变更流功能。通过变更流，MongoDB 服务器将集合中发生的更改暴露为一个事件流。Debezium 连接器会监视此流，然后将更改向下游传递。连接器首次检测到副本集时，它会检查 oplog 以获取最后记录的事务，然后执行主节点的数据库和集合的快照。连接器完成数据复制后，它会创建一个从之前读取的 oplog 位置开始的变更流。

在 MongoDB 连接器处理更改时，它会定期记录事件在 oplog 流中起源的位置。当连接器停止时，它会记录它处理的最后一个 oplog 流位置，以便在重新启动后可以从该位置继续流式传输。换句话说，连接器可以停止、升级或维护，并在稍后重新启动，并且始终能在不丢失任何事件的情况下准确地恢复到之前的位置。当然，MongoDB oplog 通常有最大容量限制，因此如果连接器停止时间过长，在连接器有机会读取它们之前，oplog 中的操作可能会被清除。在这种情况下，重新启动后，连接器会检测到丢失的 oplog 操作，执行快照，然后继续流式传输更改。

MongoDB 连接器对副本集成员身份和领导者的变化、分片集群中分片的添加或删除以及可能导致通信失败的网络问题也具有很强的容忍度。连接器始终使用副本集的主节点来流式传输更改，因此当副本集进行选举并且另一个节点成为主节点时，连接器会立即停止流式传输更改，连接到新的主节点，并使用新的主节点开始流式传输更改。类似地，如果连接器无法与副本集主节点通信，它会尝试重新连接（使用指数退避，以免使网络或副本集过载）。重新建立连接后，连接器将继续从它捕获的最后一个事件继续流式传输更改。通过这种方式，连接器可以动态地适应副本集成员的变化，并自动处理通信中断。

您可以通过在 mongodb.connection.string 中设置 readPreference 参数来指定 MongoDB 连接的读取偏好。

MongoDB 连接器支持以下 MongoDB 拓扑

为了捕获 MongoDB 中的数据，Debezium 会作为 MongoDB 用户连接到数据库。为 Debezium 创建的 MongoDB 用户帐户需要特定的数据库权限才能从数据库读取。连接器用户需要以下权限：

连接器用户可能还需要以下权限：

连接器用户必须能够从所有数据库读取，或者读取特定数据库，具体取决于连接器的 capture.scope 属性的值。根据 capture.scope 设置，为用户分配以下权限之一：

无论 capture.scope 设置如何，用户都需要运行 MongoDB hello 命令的权限。

根据您的 Debezium 环境，为了使连接器能够执行偏移量合并，您必须授予连接器用户读取 config.shards 集合的显式权限。在以下连接器环境中需要读取 config.shards 集合的权限：

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

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

Debezium MongoDB 连接器不再支持连接到分片 MongoDB 部署的 replica_set 连接。因此，由使用 replica_set 连接模式的连接器版本记录的偏移量与当前版本不兼容。

为了最小化连接模式更改的影响，并防止连接器运行不必要的快照，当连接器在升级后重新启动时，它会运行一个合并偏移量的过程。在此偏移量合并过程中，连接器将完成以下步骤来协调由早期连接器版本记录的偏移量：

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

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

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

有关更多信息，请参阅连接器配置属性表中的 snapshot.mode。

By default, a connector runs an initial snapshot operation only after it starts for the first time. Following this initial snapshot, under normal circumstances, the connector does not repeat the snapshot process. Any future change event data that the connector captures comes in through the streaming process only. (默认情况下，连接器仅在首次启动后运行初始快照操作。在初始快照之后，在正常情况下，连接器不会重复快照过程。连接器捕获的任何未来变更事件数据仅通过流式传输过程进入。)

但是，在某些情况下，连接器在初始快照期间获取的数据可能会过时、丢失或不完整。为了提供一种重新捕获集合数据的方法，Debezium 包含了一个执行临时快照的选项。您可能希望在 Debezium 环境中发生以下任何更改后执行临时快照：

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

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

临时快照信号指定要包含在快照中的集合。快照可以捕获数据库的全部内容，或者只捕获数据库中的一部分集合。

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

您可以通过向信号集合添加具有 execute-snapshot 信号类型的条目，或向 Kafka 信号主题发送信号消息来启动临时增量快照。在连接器处理消息后，它将开始快照操作。快照过程读取第一个和最后一个主键值，并将这些值作为每个集合的起始和结束点。根据集合中的条目数和配置的块大小，Debezium 将集合划分为块，然后依次对每个块进行快照。

有关更多信息，请参阅增量快照。

您可以通过向信号集合或信号主题添加具有 execute-snapshot 信号类型的条目来启动临时阻塞快照。在连接器处理消息后，它将开始快照操作。连接器会暂时停止流式传输，然后启动指定集合的快照，遵循其在初始快照期间使用的相同过程。快照完成后，连接器将恢复流式传输。

有关更多信息，请参阅阻塞快照。

To provide flexibility in managing snapshots, Debezium includes a supplementary snapshot mechanism, known as incremental snapshotting. Incremental snapshots rely on the Debezium mechanism for sending signals to a Debezium connector. Incremental snapshots are based on the DDD-3 design document. (为了在管理快照时提供灵活性，Debezium 提供了一个补充快照机制，称为增量快照。增量快照依赖于 Debezium 向 Debezium 连接器发送信号的机制。增量快照基于DDD-3 设计文档。)

在增量快照中，Debezium 不是一次性捕获数据库的全部状态（如初始快照），而是分阶段、分一系列可配置的块来捕获每个集合。您可以指定要快照的集合以及每个块的大小。块大小决定了快照在每次数据库提取操作期间收集的行数。增量快照的默认块大小为 1024 行。

随着增量快照的进行，Debezium 使用水印来跟踪其进度，维护每个捕获的集合行的记录。这种分阶段捕获数据的方法与标准的初始快照过程相比具有以下优点：

运行增量快照时，Debezium 会按主键对每个集合进行排序，然后根据配置的块大小将集合分成块。然后，它逐块地捕获每个集合中的行。对于捕获的每一行，快照都会发出一个 READ 事件。该事件代表块开始时行的值。

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

在某些情况下，流式传输过程发出的 UPDATE 或 DELETE 事件可能会乱序接收。也就是说，在快照捕获包含该行读取事件的块之前，流式传输过程可能会发出修改集合行的事件。当快照最终为该行发出相应的 READ 事件时，其值已经被覆盖。为了确保乱序的增量快照事件按正确的逻辑顺序处理，Debezium 采用了一种缓冲方案来解决冲突。仅当解决了快照事件与流式事件之间的冲突后，Debezium 才会将事件记录发送到 Kafka。

为了帮助解决乱序到达的 READ 事件与修改同一集合行的流式事件之间的冲突，Debezium 采用了一种称为快照窗口的机制。快照窗口界定了增量快照捕获指定集合块数据的间隔。在特定块的快照窗口打开之前，Debezium 会遵循其常规行为，直接将来自事务日志的事件向下游发送到目标 Kafka 主题。但从特定块的快照打开的那一刻起，直到它关闭，Debezium 会执行去重步骤来解决具有相同主键的事件之间的冲突。

For each data collection, the Debezium emits two types of events, and stores the records for them both in a single destination Kafka topic. The snapshot records that it captures directly from a table are emitted as READ operations. Meanwhile, as users continue to update records in the data collection, and the transaction log is updated to reflect each commit, Debezium emits UPDATE or DELETE operations for each change. (对于每个数据集合，Debezium 会发出两种类型的事件，并将两者的记录存储在单个目标 Kafka 主题中。它直接从表中捕获的快照记录作为 READ 操作发出。同时，随着用户继续更新数据集合中的记录，并且事务日志更新以反映每次提交，Debezium 会为每个更改发出 UPDATE 或 DELETE 操作。)

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

The connector repeats the process for each snapshot chunk. (连接器对每个快照块重复此过程。)

Currently, you can use either of the following methods to initiate an incremental snapshot (当前，您可以使用以下任一方法启动增量快照：)

要将增量快照用于分片 MongoDB 集群，必须将 incremental.snapshot.chunk.size 设置为一个足够大的值，以补偿变更流管道日益增加的复杂性。

For more advanced uses, you can fine-tune control of the snapshot by implementing one of the following interfaces (对于更高级的用途，您可以实现以下接口之一来微调快照的控制：)

To provide more flexibility in managing snapshots, Debezium includes a supplementary ad hoc snapshot mechanism, known as a blocking snapshot. Blocking snapshots rely on the Debezium mechanism for sending signals to a Debezium connector. (为了在管理快照时提供更大的灵活性，Debezium 提供了一个补充的即席快照机制，称为阻塞快照。阻塞快照依赖于 Debezium 向 Debezium 连接器发送信号的机制。)

A blocking snapshot behaves just like an initial snapshot, except that you can trigger it at run time. (阻塞快照的行为与初始快照完全相同，只是您可以随时在运行时触发它。)

You might want to run a blocking snapshot rather than use the standard initial snapshot process in the following situations (您可能希望在以下情况下运行阻塞快照，而不是使用标准的初始快照过程：)

运行阻塞快照时，Debezium 会停止流式传输，然后启动指定集合的快照，遵循其在初始快照期间使用的相同过程。快照完成后，将恢复流式传输。

You can set the following properties in the data component of a signal (您可以在信号的 data 组件中设置以下属性：)

For example (例如：)

A delay might exist between the time that you send the signal to trigger the snapshot, and the time when streaming stops and the snapshot starts. As a result of this delay, after the snapshot completes, the connector might emit some event records that duplicate records captured by the snapshot. (从您发送触发快照的信号到流式传输停止并开始快照之间可能存在延迟。由于此延迟，快照完成后，连接器可能会发出一些重复快照捕获记录的事件记录。)

在副本集的连接器任务记录偏移量后，它会使用该偏移量来确定 oplog 中应该开始流式传输更改的位置。然后，任务（根据配置）连接到副本集的主节点或连接到整个副本集的变更流，并从该位置开始流式传输更改。它处理所有创建、插入和删除操作，并将它们转换为 Debezium 更改事件。每个更改事件都包含操作在 oplog 中的位置，并且连接器会定期将其记录为最新的偏移量。偏移量记录的间隔由 offset.flush.interval.ms 控制，这是 Kafka Connect 工作者的配置属性。

当连接器正常停止时，会记录最后处理的偏移量，以便在重新启动后，连接器将从它离开的地方继续。但是，如果连接器任务意外终止，那么任务可能在其最后一次记录偏移量但尚未记录完整偏移量之前已经处理并生成了事件；重新启动后，连接器将从最后记录的偏移量开始，可能重新生成一些在崩溃前已经生成过的事件。

如前所述，连接器任务始终使用副本集的主节点从 oplog 流式传输更改，确保连接器尽可能看到最新的操作，并能够比使用从节点捕获更改具有更低的延迟。当副本集选举出新的主节点时，连接器会立即停止流式传输更改，连接到新的主节点，并从新的主节点以相同的位置开始流式传输更改。同样，如果连接器在与副本集成员通信时遇到任何问题，它会尝试重新连接（使用指数退避，以免使副本集过载），一旦连接成功，它将从上次中断的地方继续流式传输更改。通过这种方式，连接器能够动态地适应副本集成员的变化，并自动处理通信故障。

总而言之，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 工作者配置中定义 Partitioner 实现的名称来更改 Kafka 的分区逻辑。

Kafka 仅为写入单个主题分区的事件维护总顺序。通过键对事件进行分区意味着具有相同键的所有事件始终转到同一个分区。这确保了特定文档的所有事件始终具有总顺序。

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

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

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

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

启用事务元数据后，数据消息 Envelope 将会添加一个 transaction 字段。此字段提供有关每个事件的信息，形式为字段的组合：

以下是消息外观的示例：

更改事件的键包含已更改文档的键的 schema 和已更改文档的实际键。对于给定的集合，schema 和其对应的 payload 都包含一个 id 字段。此字段的值是文档的标识符，表示为从MongoDB 扩展 JSON 序列化严格模式派生的字符串。

考虑一个逻辑名称为 fulfillment、包含 inventory 数据库的副本集以及包含如下文档的 customers 集合的连接器。

捕获 customers 集合更改的每个更改事件都有相同的事件键 schema。只要 customers 集合具有之前的定义，捕获 customers 集合更改的每个更改事件都具有以下键结构。在 JSON 中，它看起来像这样：

此示例使用具有整数标识符的文档，但任何有效的 MongoDB 文档标识符都以相同方式工作，包括文档标识符。对于文档标识符，事件键的 payload.id 值是一个字符串，表示已更新文档的原始 _id 字段，作为使用严格模式的 MongoDB 扩展 JSON 序列化。下表提供了不同类型的 _id 字段如何表示的示例。

更改事件中的值比键要复杂一些。与键一样，值也包含 schema 部分和 payload 部分。schema 部分包含描述 payload 部分的 Envelope 结构的模式，包括其嵌套字段。对于创建、更新或删除数据的操作的更改事件，其值有效负载都具有信封结构。

考虑用于显示更改事件键的示例的相同样本文档

示例 customers 集合中更改的更改事件的值部分根据事件类型有所不同

以下示例显示了连接器为在 customers 集合中创建数据操作生成的更改事件的值部分：

删除更改事件中的值与同一集合的创建和更新事件具有相同的 schema 部分。删除事件中的 payload 部分包含与同一集合的创建和更新事件不同的值。特别是，删除事件既不包含 after 值也不包含 updateDescription 值。以下是 customers 集合中文档的删除事件示例：

MongoDB 连接器事件设计用于与Kafka 日志压缩配合使用。日志压缩可以删除一些较旧的消息，只要保留每个键的最新消息。这使 Kafka 可以回收存储空间，同时确保主题包含完整的数据集并且可用于重新加载基于键的状态。

所有唯一标识文档的 MongoDB 连接器事件都具有完全相同的键。当文档被删除时，删除事件值仍然可以与日志压缩一起使用，因为 Kafka 可以删除具有相同键的所有先前消息。但是，为了让 Kafka 删除具有该键的所有消息，消息值必须为 null。为了实现这一点，在 Debezium 的 MongoDB 连接器发出删除事件后，连接器会发出一个特殊的墓碑事件，该事件具有相同的键但值为 null。墓碑事件通知 Kafka 可以删除具有相同键的所有消息。

您可以将 Debezium MongoDB 连接器与MongoDB Atlas 一起使用。请注意，MongoDB Atlas 仅支持通过 SSL 进行安全连接，即连接器的 +mongodb.ssl.enabled 选项必须设置为 true。

Debezium MongoDB 连接器读取变更流以获取副本集的 oplog 数据。由于 oplog 是一个固定大小的有上限集合，如果它超过其配置的最大大小，它就会开始覆盖最早的条目。如果连接器因任何原因停止，当它重新启动时，它会尝试从最后一个 oplog 流位置恢复。但是，如果最后一个流位置已从 oplog 中删除，则根据连接器 snapshot.mode 属性中指定的值，连接器可能会启动失败，并报告无效的恢复令牌错误。如果发生故障，您必须创建一个新的连接器才能让 Debezium 继续捕获数据库中的记录。有关更多信息，请参阅如果 `snapshot.mode` 设置为 `initial`，连接器在停止很长时间后启动失败。

为了确保 oplog 保留 Debezium 恢复流式传输所需的偏移量值，您可以使用以下任一方法：

为了帮助防止与缺失 oplog 条目相关的失败，跟踪报告复制行为的指标并优化 oplog 大小以支持 Debezium 很重要。特别是，您应该监视 Oplog GB/Hour 和 Replication Oplog Window 的值。如果 Debezium 离线时间超过复制 oplog 窗口的值，并且主 oplog 的增长速度快于 Debezium 消耗条目的速度，则可能会导致连接器失败。

有关如何监视这些指标的信息，请参阅MongoDB 文档。

最好将最大 oplog 大小设置为基于 oplog 的预期每小时增长量（Oplog GB/Hour）乘以解决 Debezium 故障可能需要的时间。

即：

Oplog GB/Hour X 平均 Debezium 故障响应时间

例如，如果 oplog 大小限制设置为 1GB，并且 oplog 每小时增长 3GB，则 oplog 条目每小时清除三次。如果 Debezium 在此期间发生故障，其最后 oplog 位置很可能会被删除。

如果 oplog 以 3GB/小时的速度增长，并且 Debezium 离线两个小时，那么您会将 oplog 大小设置为 3GB/小时 X 2 小时，即 6GB。

以下是一个连接器实例的配置示例，该实例用于捕获 192.168.99.100 的 27017 端口上名为 rs0 的 MongoDB 副本集的数据，我们逻辑上将其命名为 fullfillment。通常，您会在 JSON 文件中通过设置连接器可用的配置属性来配置 Debezium MongoDB 连接器。

您可以选择为特定的 MongoDB 副本集或分片集群生成事件。可选地，您可以过滤掉不需要的集合。

有关 Debezium MongoDB 连接器可设置的完整配置属性列表，请参阅 MongoDB 连接器配置属性。

您可以使用 POST 命令将此配置发送到正在运行的 Kafka Connect 服务。该服务会记录配置并启动一个连接器任务，该任务执行以下操作：

要开始运行 Debezium MongoDB 连接器，请创建连接器配置，并将配置添加到您的 Kafka Connect 集群。

连接器启动后，它将完成以下操作

Debezium MongoDB 连接器具有许多配置属性，您可以使用它们来实现连接器为您的应用程序实现正确的行为。许多属性都有默认值。有关属性的信息组织如下

以下配置属性是必需的，除非有默认值可用。

Debezium 连接器通过连接器的 MBean 名称公开指标。这些指标特定于每个连接器实例，提供有关连接器快照、流式传输和模式历史记录进程行为的数据。

默认情况下，当您部署正确配置的连接器时，Debezium 会为不同的连接器指标生成唯一的 MBean 名称。要查看连接器进程的指标，请将您的可观察性堆栈配置为监视其 MBean。但这些默认 MBean 名称取决于连接器配置；配置更改可能导致 MBean 名称发生更改。在这种情况下，更改 MBean 名称会破坏连接器实例和 MBean 之间的链接，从而中断监控活动。您必须重新配置可观察性堆栈以使用新的 MBean 名称才能恢复监控。

为了防止因 MBean 名称更改而导致的监控中断，您可以配置自定义指标标签。通过将 custom.metric.tags 属性添加到连接器配置中来配置自定义指标。该属性接受键值对，其中每个键代表 MBean 对象名称的标签，相应的值代表该标签的值。例如：k1=v1,k2=v2。Debezium 将指定的标签附加到连接器的 MBean 名称。

配置连接器的 custom.metric.tags 属性后，您可以配置可观察性堆栈以检索与指定标签关联的指标。然后，可观察性堆栈使用指定的标签而不是可变的 MBean 名称来唯一标识连接器。之后，如果 Debezium 重新定义了其如何构造 MBean 名称，或者连接器配置中的 topic.prefix 发生更改，指标收集将不会中断，因为指标抓取任务使用指定的标签模式来标识连接器。

使用自定义标签的另一个好处是，您可以使用反映数据管道架构的标签，以便以适合您运营需求的方式组织指标。例如，您可以指定带有值的标签来声明连接器活动的类型、应用程序上下文或数据源，例如 db1-streaming-for-application-abc。如果您指定多个键值对，所有指定的对都将附加到连接器的 MBean 名称。

以下示例说明了标签如何修改默认 MBean 名称。

MBean 是 debezium.mongodb:type=connector-metrics,context=snapshot,server=<topic.prefix>,task=<task.id>。

除非快照操作正在活动中，或者自上次连接器启动以来发生过快照，否则快照指标不会暴露。

下表列出了可用的快照指标。

Debezium MongoDB 连接器还提供以下自定义快照指标

MBean 是 debezium.mongodb:type=connector-metrics,context=streaming,server=<topic.prefix>,task=<task.id>。

下表列出了可用的流式传输指标。

Debezium MongoDB 连接器还提供以下自定义流式传输指标

在以下情况下，连接器在尝试启动时会失败，在日志中报告错误或异常，并停止运行：

发生故障后，连接器将尝试使用指数退避机制重新连接。您可以配置最大重连尝试次数。

在这些情况下，错误将包含有关问题的更多详细信息，并可能提供建议的解决方法。当配置已更正或 MongoDB 问题已解决后，可以重新启动连接器。

一旦连接器运行，如果任何 MongoDB 副本集的主节点变得不可用或无法访问，连接器将反复尝试重新连接到主节点，使用指数退避机制以防止网络或服务器过载。如果主节点在可配置的连接尝试次数后仍不可用，连接器将失败。

尝试重新连接由三个属性控制

每次延迟是前一次延迟的两倍，直到达到最大延迟。考虑到默认值，下表显示了每次连接失败尝试的延迟以及失败前的累计总时间。

停止很长时间后停止的连接器无法启动，并报告以下异常

前面的异常表明对应于连接器恢复令牌的条目已不再存在于 oplog 中。由于 oplog 不再包含连接器处理的最后一个偏移量，因此连接器无法恢复流式传输。

您可以使用以下任一选项从故障中恢复

为防止与丢失的恢复令牌相关的故障，请优化 oplog 配置。

如果 Kafka Connect 在分布式模式下运行，并且 Kafka Connect 进程正常停止，那么在进程关闭之前，Kafka Connect 会将该进程的所有连接器任务迁移到该组中的另一个 Kafka Connect 进程，新连接器任务将从之前任务中断的地方继续。在连接器任务正常停止并在新进程上重新启动期间会有短暂的处理延迟。

如果组中只有一个进程，并且该进程正常停止，那么 Kafka Connect 将停止连接器并记录每个副本集的最后一个偏移量。重新启动后，副本集任务将从它们停止的地方继续。

如果 Kafka Connector 进程意外停止，那么它正在运行的任何连接器任务将在未记录其最近处理的偏移量的情况下终止。当 Kafka Connect 在分布式模式下运行时，它将在其他进程上重新启动这些连接器任务。但是，MongoDB 连接器将从早期进程记录的最后一个偏移量恢复，这意味着新的替换任务可能会生成一些在崩溃前刚刚处理过的相同变更事件。重复事件的数量取决于偏移量刷新周期和崩溃前的数据更改量。

当连接器生成变更事件时，Kafka Connect 框架使用 Kafka producer API 将这些事件记录到 Kafka 中。Kafka Connect 还将定期记录这些变更事件中出现的最后一个偏移量，频率由您在 Kafka Connect 工作进程配置中指定。如果 Kafka brokers 不可用，运行连接器的 Kafka Connect 工作进程将简单地反复尝试重新连接到 Kafka brokers。换句话说，连接器任务将暂停，直到可以重新建立连接，此时连接器将从中断的地方继续。

如果连接器被正常停止，用户可以继续在副本集成员上执行操作。连接器离线期间发生的更改将继续记录在 MongoDB 的 oplog 中。在大多数情况下，连接器重新启动后，它会读取 oplog 中的偏移量值，以确定它为每个副本集流式传输的最后一个操作，然后从该点恢复流式传输。重新启动后，连接器离线期间发生的数据库操作将像往常一样发送到 Kafka，并在一段时间后，连接器将赶上数据库。连接器赶上数据库所需的时间取决于 Kafka 的功能和性能以及数据库中发生的更改量。

但是，如果连接器停止的时间过长，可能会发生 MongoDB 在连接器不活动期间清除 oplog 的情况，导致丢失有关连接器最后一个位置的信息。连接器重新启动后，由于 oplog 中不再包含标记连接器处理的最后一个操作的先前偏移量值，因此连接器无法恢复流式传输。连接器也无法执行快照，尽管通常当 snapshot.mode 属性设置为 initial 且没有偏移量值时，它会这样做。在这种情况下，会出现不匹配，因为 oplog 不包含先前偏移量的值，但该偏移量值存在于连接器的内部 Kafka 偏移量主题中。这将导致错误，连接器将失败。

要从故障中恢复，请删除失败的连接器，然后创建一个具有相同配置但名称不同的新连接器。启动新连接器时，它将执行快照以摄取数据库的状态，然后恢复流式传输。

在某些故障情况下，MongoDB 可能会丢失提交，导致 MongoDB 连接器无法捕获丢失的更改。例如，如果主节点在应用更改并将其记录到 oplog 后突然崩溃，oplog 可能会在辅助节点能够读取其内容之前变得不可用。结果，被选为新主节点的辅助节点可能缺少其 oplog 中的最新更改。

目前，没有办法防止 MongoDB 中出现此副作用。