Debezium MongoDB 连接器不执行初始快照

Question

我正在将 MongoDB atlas 与分片副本集集群一起使用，Debezium MongoDB connector 如文档中所述。

这是我当前配置的样子（运行独立设置）：

name=dev-mongodb
connector.class=io.debezium.connector.mongodb.MongoDbConnector
tasks.max=4
mongodb.hosts=<some-url>.mongodb.net:27017
mongodb.name=mongodb
mongodb.user=<admin_user>
mongodb.password=<admin_user_pw>
database.include.list=<list_of_databases>
database.history.kafka.bootstrap.servers=<list_of_aws_msk_brokers>
database.history.kafka.topic=mongodb.history
include.schema.changes=true
mongodb.ssl.enabled=true

我可以在 kafka 主题中接收 CDC 事件，但从未制作过文档描述的初始快照。我尝试过使用不同的mongodb.name，导致创建和使用的主题集完全不同，但结果相同。

MongoDB oplog 有大约 2M 行，kafka 主题总共只有几千条消息。

进一步挖掘，似乎连接器记录了 oplog 的最后一个位置的偏移量。是否可以重置此偏移量？

Answer 1

在我看来，您在多个部署中使用了相同的连接器名称，这意味着尽管更改了配置并尝试重置连接器的状态，但它仍会继续查找先前的偏移量并恢复 oplog 位置。

有两种选择：

• 使用完全不同的连接器名称创建一个新连接器。
• 手动清除连接器的偏移量

很多用户更喜欢第一个选项，因为它最简单。 Kafka 根据连接器的 name 记录连接器的偏移量，因此只需调整连接器的名称即可告诉 Kafka 该连接器是全新的，它不会找到任何要恢复的持久偏移量。

第二个选项有点复杂，因为您需要首先找到存储偏移量的 Kafka 主题，通常默认为 connect-offsets，但可以被覆盖。了解主题后，您应该关闭所有使用该主题的连接器。如果您在连接器使用它时调整此主题，可能会导致意外行为。

使用 Kafka 提供的 kafkacat 工具，您需要运行以下命令，它假定默认的连接偏移量主题名称，因此请相应地进行调整：

$ kafkacat -b localhost:9092 -t connect-offsets -C -f '\nKey (%K bytes): %k
  Value (%S bytes): %s
  Timestamp: %T
  Partition: %p
  Offset: %o\n'

这将产生一些输出，重要的是要注意“密钥”和“分区”。为了重置偏移量，您需要使用正确的“Key”和“Partition”值有效地将NULL（或墓碑）写入主题。

假设上面提供了这个输出：

% Reached end of topic connect-offsets [0] at offset 0
% Reached end of topic connect-offsets [1] at offset 0
[…]
Key (52 bytes): ["source-file-01",{"filename":"/data/testdata.txt"}]
Value (15 bytes): {"position":87}
Timestamp: 1565859303551
Partition: 20
Offset: 0
[…]

你会想要执行以下命令：

$ echo '["source-file-01",{"filename":"/data/testdata.txt"}]#' | \
    kafkacat -b localhost:9092 -t connect-offsets -P -Z -K# -p 20

在 echo 语句中，我们指定键后跟由 kafkacat 参数 -K# 定义的键分隔符 # 和 -Z 选项，即发送一个空值作为 NULL。 -p 参数是指定分区的位置，正确设置键和分区很重要。

完成此操作后，您可以安全地重新启动使用该偏移主题的连接器，您应该会看到连接器的行为就像一个全新的部署。

请注意，如果您正在使用使用数据库历史主题（如 MySQL、SQL Server 或 Oracle）的连接器，则还需要清除数据库历史主题。

然而，正如我之前所说，使用新名称重新部署连接器会更简单，以避免需要执行所有 kafka 主题魔法才能达到相同的结果。