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 可以自动创建主题。

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

您可以通过向信号集合发送 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 事件的块之前，发出修改集合行的事件。当快照最终发出该行的相应 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 集合中创建数据的操作生成的更改事件的value 部分。

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

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

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

您可以将 Debezium for 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。

以下是一个连接器实例的配置示例，该实例从端口 27017 上的 MongoDB 副本集 rs0（位于 192.168.99.100）捕获数据，并将其逻辑命名为 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 worker 配置中指定。如果 Kafka 代理变得不可用，运行连接器的 Kafka Connect worker 进程将简单地反复尝试重新连接到 Kafka 代理。换句话说，连接器任务将简单地暂停，直到可以重新建立连接，届时连接器将从它们中断的地方继续。

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

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

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

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

目前，没有办法防止 MongoDB 中的这种副作用。