Kakfa-MongoDB 源连接器流中缺少恢复令牌答案

【问题标题】：Resume token missing in Kakfa-MongoDB source connector streamKakfa-MongoDB 源连接器流中缺少恢复令牌
【发布时间】：2020-12-27 00:40:38
【问题描述】：

我正在使用 Confluent Platform v5.4、Mongo DB 3.6 版和 Kafka Mongo DB 连接器。我已将 Kafka Mongo DB 源连接器配置为在 Mongo DB 集合中创建新记录时将数据从 Mongo DB 推送到 Kafka 主题。我已将我的 Kafka 代理配置为将数据日志保留 1 天，之前设置为默认配置 7 天。现在我的问题是，我在 Mongo DB 日志中反复出现这样的错误 -

2020-09-08T14:58:34.500+0000 I COMMAND  [conn66] command local.oplog.rs command: aggregate { aggregate: "topic_1", pipeline: [ { $changeStream: { fullDocument: "default", resumeAfter: { _data: BinData(0, 825F50DCCE00000008463C5F6964003C35363962633732642D386430352D343034622D616262362D64363632656136663464303000005A1004B31FA99AA5C141CB9EE9AA845877B80D04) } } } ], cursor: {}, $db: "ctc", $clusterTime: { clusterTime: Timestamp(1599577091, 6), signature: { hash: BinData(0, 6550821FD90928ABC3A2DFE066FC34E824A83762), keyId: 6847473605622628353 } }, lsid: { id: UUID("2a2e5878-4743-4e83-a4fd-eed68b5efe02") }, $readPreference: { mode: "primaryPreferred" } } planSummary: COLLSCAN exception: resume of change stream was not possible, as the resume token was not found. {_data: BinData(0, "825F579ACE00000008463C5F6964003C31313632333863352D316561612D343961652D613437632D30366263336131333436313900005A1004B31FA99AA5C141CB9EE9AA845877B80D04")} code:ChangeStreamFatalError numYields:8932 reslen:455 locks:{ Global: { acquireCount: { r: 17872 }, acquireWaitCount: { r: 97 }, timeAcquiringMicros: { r: 350686 } }, Database: { acquireCount: { r: 8936 } }, Collection: { acquireCount: { r: 1 } }, oplog: { acquireCount: { r: 8935 } } } protocol:op_msg 22198ms

据我了解，在 Kafka 主题和 Mongo DB 的 Oplog 之间，有一个共享的恢复令牌以保持数据流正常运行。但是有时 Oplog 会被刷掉（因为 Oplog 的大小只能是分配给 Mongo DB 的内存的一定比例）或者 Kafka 主题对应的数据被删除，从而导致流的破裂。

我的问题是 - 如何避免流中断？如何确保恢复令牌始终存在于 Oplog 和 Kafka 主题中？有什么方法可以让我从可用的地方手动获取恢复令牌并在丢失的地方更新恢复令牌？

【问题讨论】：

标签： mongodb apache-kafka apache-kafka-connect confluent-platform mongodb-kafka-connector

【解决方案1】：

在 4.4 兼容的驱动程序中进行了更改，以处理长时间没有任何更改时从 oplog 末尾掉落的更改流恢复令牌。 Ruby 驱动程序的用户端描述为here；根据需要适应 Java。关键是您需要使用tryNext 并从更改流中读取恢复令牌，而不是从返回的更改文档中读取。

如果您正在执行上述所有操作，您需要：

增加 oplog 窗口并保留更多 oplog，或
更快地处理更改以保持在 oplog 窗口内。

【讨论】：