mongoDB 基于flink-cdc的实时读Connector实现

[baisui1981]

https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2/docs/connectors/flink-sources/mongodb-cdc/#startup-reading-position

[baisui1981]

mongod CDC 增量任务启动报错，处理：

The $changeStream stage is only supported on replica sets

• 详细
  flink 增量任务启动报以下错误：Caused by: com.mongodb.MongoCommandException: Command failed with error 40573 (Location40573): 'The $changeStream stage is only supported on replica sets' on server 192.168.28.201:27017. The full response is {"ok": 0.0, "errmsg": "The $changeStream stage is only supported on replica sets", "code": 40573, "codeName": "Location40573"}
• 解决：

这个错误表明你尝试在不是副本集（replica set）的 MongoDB 实例上使用 $changeStream 聚合管道阶段。MongoDB 的变更流（Change Streams）特性依赖于副本集来工作，因为它需要确保数据的持久性和可用性，以及提供必要的操作日志（oplog）来跟踪数据库的变化。

要解决这个问题，你需要将你的 MongoDB 实例配置为一个副本集。以下是设置 MongoDB 副本集的基本步骤：

1. 单节点副本集
   如果你只需要一个单节点的副本集用于测试或开发环境，你可以按照以下步骤进行设置：

   启动 MongoDB：确保你的 MongoDB 服务正在运行，并且可以通过提供的地址和端口访问。
   初始化副本集：连接到你的 MongoDB 实例并执行以下命令以初始化一个单节点的副本集。假设你已经启动了 MongoDB 并.
   且它正在监听 192.168.28.201:27017。

mongo --host 192.168.28.201 --port 27017

然后，在 mongo shell 中输入：

rs.initiate()

这将会把当前的 MongoDB 实例配置为一个名为 rs0 的副本集成员。

验证副本集状态：你可以通过以下命令来检查副本集的状态：

  rs.status()

2. 多节点副本集
   对于生产环境，建议设置一个多节点副本集以增加高可用性和容错能力。这涉及到多个步骤，包括配置多个 MongoDB 实例，并将它们添加到同一个副本集中。这里不详细展开多节点副本集的设置过程，但基本思路是类似的，只是需要在多个服务器上重复上述步骤，并在初始化时指定所有成员。

注意事项

• 如果你在使用 Docker 或 Kubernetes 等容器化平台，请确保正确地设置了网络和服务发现，以便各副本集成员能够相互通信。
• 对于生产环境，强烈建议配置至少三个节点的副本集，以确保选举主节点的能力和更好的故障转移机制。
• 在配置副本集之前，请查阅最新的 MongoDB 文档，因为配置选项可能会随着版本更新而有所变化。
  完成以上步骤后，你应该能够使用变更流功能了。如果你已经在运行一个副本集，但是仍然遇到相同的错误，请检查 MongoDB 日志文件，确认副本集是否正常工作，并且所有成员都健康在线。

MongoServerError[NoReplicationEnabled]: This node was not started with replication enabled.

执行 rs.initiate(); 报以上错误。这个错误信息 MongoServerError[NoReplicationEnabled]: This node was not started with replication enabled. 表明 MongoDB 实例并没有配置为副本集成员，因此无法使用需要副本集功能的特性，如变更流（Change Streams）。

要解决这个问题，你需要确保 MongoDB 服务是以副本集模式启动的。下面是如何在启动 MongoDB 时启用复制功能的指导：

• 启用复制功能

  1. 编辑配置文件：
     找到你的 MongoDB 配置文件（通常是 mongod.conf），并确保它包含以下内容来设置副本集名称。假设你要创建一个名为 rs0 的副本集：

     replication:
       replSetName: "rs0"

  2. 初始化副本集：
     如果你已经在运行 MongoDB，请先停止它。
     然后使用带有上述配置的命令重新启动 MongoDB。
     连接到 MongoDB，并通过 mongo shell 初始化副本集：

     mongo --host 192.168.28.201 --port 27017

     在 mongo shell 中执行以下命令：

     rs.initiate()

  3. 验证副本集状态：
     使用 rs.status() 命令检查副本集的状态，确认一切正常。

• 对于 Docker 用户
  如果你是通过 Docker 容器运行 MongoDB，确保在启动容器时传递了正确的参数以启用复制功能。例如：

        docker run -d \
        --name mongodb  \
        -p 27017:27017 \
        -v /opt/misc/mongo/data:/data/db \
        -v /opt/misc/mongo/conf:/data/conf \
        -v /opt/misc/mongo/log:/data/log \
        -e MONGO_INITDB_ROOT_USERNAME=admin \
        -e MONGO_INITDB_ROOT_PASSWORD=123456 \
        -e MONGO_INITDB_REPLICA_SET=rs0 \
        --privileged=true \
        --restart always \
        mongo:6.0.14 mongod --replSet rs0

在这个例子中，MONGO_INITDB_REPLICA_SET 环境变量用于指定副本集的名字，并且 --replSet 参数告诉 MongoDB 服务以副本集模式启动。

• 注意事项
  1. 如果你在生产环境中运行 MongoDB，建议至少配置三个节点的副本集以确保高可用性和容错能力。
  2. 确保所有副本集成员之间的网络连接稳定，这对于选举主节点和维护数据一致性非常重要。
  3. 如果你正在使用旧版本的 MongoDB，请参考对应版本的官方文档来获取最准确的配置说明。
  完成这些步骤之后，你的 MongoDB 应该已经正确地配置为一个副本集成员，你可以再次尝试启动 Flink 增量任务，应该不会再遇到 NoReplicationEnabled 错误。