ClassNotFoundException: com.mongodb.ConnectionString 用于 Apache Kafka Mongodb 连接器

Question

我正在我的 Windows 机器上配置一个 Kafka Mongodb 接收器连接器。

我的 connect-standalone.properties 文件有

plugin.path=E:/Tools/kafka_2.12-2.4.0/plugins

我的 MongoSinkConnector.properties 文件有

name=mongo-sink
topics=first_topic
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1
key.ignore=true

# Specific global MongoDB Sink Connector configuration
connection.uri=mongodb://localhost:27017,mongo1:27017,mongo2:27017,mongo3:27017
database=test_kafka
collection=transactions
max.num.retries=3
retries.defer.timeout=5000
type.name=kafka-connect

在 E:/Tools/kafka_2.12-2.4.0/plugins 文件夹中，我有 mongo-kafka-connect-1.0.1.jar 文件。

命令

bin\windows\connect-standalone config\connect-standalone.properties config\MongoSinkConnector.properties

我得到的错误是

[2020-03-23 04:04:12,376] ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone)
java.lang.NoClassDefFoundError: com/mongodb/ConnectionString
        at com.mongodb.kafka.connect.sink.MongoSinkConfig.createConfigDef(MongoSinkConfig.java:140)
        at com.mongodb.kafka.connect.sink.MongoSinkConfig.<clinit>(MongoSinkConfig.java:78)
        at com.mongodb.kafka.connect.MongoSinkConnector.config(MongoSinkConnector.java:62)
        at org.apache.kafka.connect.connector.Connector.validate(Connector.java:129)
        at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:313)
        at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig(StandaloneHerder.java:194)
        at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:115)
Caused by: java.lang.ClassNotFoundException: com.mongodb.ConnectionString
        at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:588)
        at org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:104)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)

我应该在插件文件夹中放置哪些其他 jar 文件和/或我必须进行配置更改？

更新 1 我已经将 mongodb-driver-core-4.0.1 和 bson-4.0.1 jar 文件也放在了 plugins 文件夹中，但有同样的错误。

Answer 1

最后，我可以让 mongo-kafka-connector 在 Windows 上工作。

这对我有用： Kafka安装文件夹为E:\Tools\kafka_2.12-2.4.0

E:\Tools\kafka_2.12-2.4.0\plugins 有 mongo-kafka-1.0.1-all.jar 文件。

我从https://www.confluent.io/hub/mongodb/kafka-connect-mongodb下载了这个 点击左侧蓝色的下载按钮，获取mongodb-kafka-connect-mongodb-1.0.1.zip文件。

在 zip 文件的 etc 文件夹中还有文件 MongoSinkConnector.properties。 将其移至 kafka_installation_folder\plugins

我的connect-standalone.propertie文件包含以下条目：

bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=false
value.converter.schemas.enable=false

offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=E:/Tools/kafka_2.12-2.4.0/plugins/mongo-kafka-1.0.1-all.jar

我的 MongoSinkConnector.properties 文件包含以下条目

name=mongo-sink
topics=topic1,topic2
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1

connection.uri=mongodb://localhost:27017,localhost:27017,localhost:27017
database=test_kafka
collection=transactions
max.num.retries=3
retries.defer.timeout=5000

field.renamer.mapping=[]
field.renamer.regex=[]

max.batch.size = 0
rate.limiting.timeout=0
rate.limiting.every.n=0

如何运行

在三个控制台启动mongodb、zookeeper、kafka server。

在第 4 个控制台中，启动 Kafka 连接 --

bin\windows\connect-standalone config\connect-standalone.properties config\MongoSinkConnector.properties

在第 5 个控制台中，将消息发送到一个主题（我为 topic1 做过）

bin\windows\kafka-console-producer --broker-list localhost:9092 --topic topic1

>{“你好”：1}

>{“蒙古”：2}

>{“世界”：3}

打开一个 mongo 客户端并检查您的数据库/集合。您将看到这三个消息。