问题标签 [mongodb-kafka-connector]

我正在研究 debezium mongodb 源连接器。我可以通过将 kafka 引导服务器地址作为远程机器（部署在 Kubernetes 中）和远程 MongoDB url 以分布式模式在本地机器上运行连接器吗？

我试过这个，我看到连接器成功启动，没有错误，只有几个警告，但没有数据从 mongodb 流出。

使用以下命令运行连接器

如果不是我怎么能做到这一点，我不想像大多数教程建议的那样安装本地 kafka 或 mondoDB。我想为此使用我们的测试服务器。

遵循以下教程 ：https ://medium.com/tech-that-works/cloud-kafka-connector-for-mongodb-source-8b525b779772

以下是有关连接器工作正常问题的更多详细信息，我在连接器日志末尾看到以下行

我还在 /etc/kafka/connect-mongodb-source.properties 中定义了 MongoDB 配置，如下所示

但是数据并没有在 MongoDB 和 Kafka 之间流动。我还为此 Kafka-MongoDB Debezium 连接器发布了单独的问题：分布式模式

任何指针都适用

下面是我在 /etc/kafka/connect-mongodb-source.properties 中的 MongoDB 配置

但低于错误

在独立模式下运行连接器。

我在 debezium-debezium-connector-mongodb-1.0.0/debezium-connector-mongodb-1.0.0.Final.jar 下有 mongodbconnector 类

类路径也设置如下

使用插件路径，我看到它能够注册和加载所有必需的插件..

但最后还是同样的问题

>curl localhost:8083/connector-plugins 的输出

提前致谢

我正在使用 Kafka-MongoDB-Connect 在哪里

1. 来源是 Debezium 连接器
2. Sink 是 MongoDB 的 Kafka-MongoDB-Connector

我可以对 Source 和 Sink Connector 使用相同的主题和 MongoDB 集合吗？它会导致消息无限循环吗？

使用连接器将数据从 mongodb 导入 kafka 时，https://github.com/mongodb/mongo-kafka会抛出java.lang.IllegalStateException: Queue full.

我使用默认设置copy.existing.queue.size16000 和copy.existing: true. 我应该设置什么值？集合大小为10G。

环境：

堆栈跟踪： org.apache.kafka.connect.errors.ConnectException: java.lang.IllegalStateException: Queue full\n\tat com.mongodb.kafka.connect.source.MongoCopyDataManager.poll(MongoCopyDataManager.java:95)\n\tat com.mongodb.kafka.connect.source.MongoSourceTask.getNextDocument(MongoSourceTask.java:301)\n\tat com.mongodb.kafka.connect.source.MongoSourceTask.poll(MongoSourceTask.java:154)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:265)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)\n\tat java.base\/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)\n\tat java.base\/java.util.concurrent.FutureTask.run(Unknown Source)\n\tat java.base\/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)\n\tat java.base\/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)\n\tat java.base\/java.lang.Thread.run(Unknown Source)\nCaused by: java.lang.IllegalStateException: Queue full\n\tat java.base\/java.util.AbstractQueue.add(Unknown Source)\n\tat java.base\/java.util.concurrent.ArrayBlockingQueue.add(Unknown Source)\n\tat com.mongodb.client.internal.Java8ForEachHelper.forEach(Java8ForEachHelper.java:30)\n\tat com.mongodb.client.internal.Java8AggregateIterableImpl.forEach(Java8AggregateIterableImpl.java:54)\n\tat com.mongodb.kafka.connect.source.MongoCopyDataManager.copyDataFrom(MongoCopyDataManager.java:123)\n\tat com.mongodb.kafka.connect.source.MongoCopyDataManager.lambda$new$0(MongoCopyDataManager.java:87)\n\t... 5 more

我在独立的 kafka 节点上使用 mongo-kafka 连接器。是否有任何优雅的方法（不使用 kill 方法）来杀死 mongo-kafka 连接器进程。

我正在尝试将 MongoDB 作为源连接到 Kafka 连接服务器，但是当我运行 CURL POST 命令来注册 Mongo 源连接器时。我成功验证了 kafka-connect 连接器插件中连接器的存在，所以我不明白为什么会出现这些错误

和

这是我的 docker-compose 配置

我需要从 Kafka Topic 和 Sink 监听事件到 MongoDB 中的集合。该消息包含一个带有 id 属性的嵌套对象，如上例所示。

我正在尝试使用 RegExp 将此嵌套 id 重命名为 _id

配置/验证的结果是500 Internal Server Error，带有该消息：

我错过了什么或者是一个问题？

我有一个要求，我必须捕获 MongoDB ChangeStream（插入/更新等）事件并采取一些措施（保存到 OracleDB）。我想到了这个似乎很好的设计：

"MongoDB-ChangeStream" --> "MongoDB Source Connector for Apache Kafka" --> "Kafka Broker-Topic" --> "Java Service" --> OracleDB

我的问题是，我使用的是 MongoDB-4.0，并且在 MongoDB-4.2 中引入了“MongoDB Source Connector for Apache Kafka”。我还能在 MongoDB-4.0 中使用“MongoDB Source Connector for Apache Kafka”吗？

MongoDB Source Connector for Apache Kafka - 官方页面： https ://www.confluent.io/hub/mongodb/kafka-connect-mongodb

“MongoDB Source Connector for Apache Kafka”官方页面并没有谈及MongoDB版本方面的任何先决条件。

我的第二个问题，这个（Apache Kafka 的 MongoDB 源连接器）可以在 Windows 上设置吗？

感谢任何帮助。

我正在尝试将 monodb 连接到 kafka，因此每当将新文档插入 mongo 时，我都会将数据发送到 kafka 主题。所以我使用融合集线器安装了 mongodb 连接器，配置 MongoSourceConnector.properties 如下：

重新启动融合 kafka 并将一些新文档插入到 mongo 中，但我在 kafka 上什么也没看到，没有创建新主题，似乎没有任何效果！我在这里错过了什么吗？还有什么我应该配置/运行的吗？

谢谢 ：）

我有一个已经在运行的生产部署的 Kafka-Cluster 并且主题为“ existing-topic ”。我正在使用来自 Debezium 的 MongoDB-Source-Connector。

在这里，我想要的只是将 CDC 事件直接推送到主题“ existing-topic ”，以便已经在收听该主题的消费者将处理它。

我没有找到任何资源可以这样做，但是提到该主题是按以下格式创建的-

“如果您的 mongodb.name 参数为 A，数据库名称为 B，集合名称为 C，则数据库 A 和集合 C 中的数据将加载到主题 ABC 下”

我可以将主题更改为“现有主题”并将事件推送到它吗？