mongo kafka 连接源答案

【问题标题】：mongo kafka connect sourcemongo kafka 连接源
【发布时间】：2020-05-08 12:08:28
【问题描述】：

我正在使用 kafka connect 来从 mongo 读取数据并将它们写入 kafka 主题。

我正在使用 mongo kafka 源连接器。

我收到以下错误：

ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:115)
java.lang.NoClassDefFoundError: com/mongodb/ConnectionString
    at com.mongodb.kafka.connect.source.MongoSourceConfig.createConfigDef(MongoSourceConfig.java:209)
    at com.mongodb.kafka.connect.source.MongoSourceConfig.<clinit>(MongoSourceConfig.java:138)
    at com.mongodb.kafka.connect.MongoSourceConnector.config(MongoSourceConnector.java:56)
    at org.apache.kafka.connect.connector.Connector.validate(Connector.java:129)
    at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:282)
    at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig(StandaloneHerder.java:188)
    at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:109)
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)
    ... 7 more

jar 中似乎有 amising 类。为了得到罐子，我使用了两种不同的方法，但我得到了同样的错误。首先我使用了来自：maven 存储库的下载，然后我从 github repo 克隆了源代码，然后我自己构建了 jar。我将 jar 推到 plugins.path。当我解压缩生成的 jar 并浏览 calsses 时，我找不到提到的类：com.mongodb.ConnectionString

我使用了以下配置文件

worker.properties：

 rest.port=18083

# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
# (connectors, converters, transformations). The list should consist of top level directories that include 
# any combination of: 
# a) directories immediately containing jars with plugins and their dependencies
# b) uber-jars with plugins and their dependencies
# c) directories immediately containing the package directory structure of classes of plugins and their dependencies
# Note: symlinks will be followed to discover dependencies or plugins.
# Examples: 
# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
plugin.path=/usr/share/java/plugins

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
bootstrap.servers=127.0.0.1:9092

mongo-connector.properties：

name=mongo
tasks.max=1
connector.class =com.mongodb.kafka.connect.MongoSourceConnector
database=
collection=alerts
key.converter = org.apache.kafka.connect.storage.StringConverter
value.converter = org.apache.kafka.connect.storage.StringConverter

topic.prefix=someprefix
poll.max.batch.size=1000
poll.await.time.ms=5000

# Change stream options
pipeline=[]
batch.size=0
change.stream.full.document=updateLookup

然后我通过以下命令启动了连接器：

/usr/local/kafka/bin/connect-standalone.sh worker.properties mongo-connector.properties

知道如何解决这个问题

【问题讨论】：

标签： mongodb apache-kafka apache-kafka-connect

【解决方案1】：

您必须将连接器的 JAR 文件放在 plugin.path 下，在您的情况下是 /usr/share/java/plugins。

说明已经存在于Confluent's documentation：

Kafka Connect 插件是：

一个包含插件及其所有类文件的 uber JAR 单个 JAR 文件中的第三方依赖项；或目录包含插件及其 JAR 文件的文件系统第三方依赖。但是，插件永远不应该包含任何 Kafka Connect 运行时提供的库。

Kafka Connect 使用其插件路径查找插件，这是一个 Kafka Connect 中定义的以逗号分隔的目录列表工人配置。要安装插件，请放置插件目录或 uber JAR（或解析为其中之一的符号链接）插件路径上列出的目录，或将插件路径更新为包括包含插件的目录的绝对路径。

【讨论】：

@scalacode 在进行这些更改后，您是否重新启动了 Kafka Connect，以便它能够重新读取 plugins.path 内容并加载连接器？
我更改了连接器的名称，但这也不起作用
@scalacode 您的连接器的 jar 文件在哪里？
它现在可以工作了，我从 confluent 中拿走了罐子，而不是 maven 中的官方罐子

【解决方案2】：

我正在创建这个答案，因为我花了一些时间来找出解决方案，正如 scalacode 所指出的，最简单的解决方案是从 confluent 下载 jar，而不是从 maven。

https://www.confluent.io/hub/mongodb/kafka-connect-mongodb

【讨论】：