Debezium Kafka 连接器 mongodb：将 kafka 连接器连接到 mongodb 时出错

Posted 2023-03-23

【中文标题】Debezium Kafka 连接器 mongodb：将 kafka 连接器连接到 mongodb 时出错

【英文标题】：Debezium Kafka connector mongodb : Error connecting kafka connector to mongodb

【发布时间】：2020-04-27 17:21:41

【问题描述】：

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

name=mongodb-source-connector 
connector.class=io.debezium.connector.mongodb.MongoDbConnector 
mongodb.hosts=/remoteserveraddress:27017 
mongodb.name=mongo_conn 
initial.sync.max.threads=1 
tasks.max=1

但低于错误

 ERROR Plugin class loader for connector: 'io.debezium.connector.mongodb.MongoDbConnector' was not found. Returning: org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader@5a058be5 (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:165)

以独立模式运行连接器。

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

classpath也设置如下

#for CLASSPATH
CLASSPATH=/Users/111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/*
export CLASSPATH
PATH=$PATH:/usr/local/sbin
export PATH

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

[2020-01-10 08:14:07,916] INFO Scanning for plugin classes. This might take a moment ... (org.apache.kafka.connect.cli.ConnectStandalone:78)
[2020-01-10 08:14:07,942] INFO Loading plugin from: /Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/mongodb-driver-3.11.1.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:241)
[2020-01-10 08:14:08,082] INFO Registered loader: PluginClassLoaderpluginLocation=file:/Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/mongodb-driver-3.11.1.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:264)
[2020-01-10 08:14:08,083] INFO Added plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,083] INFO Added plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,083] INFO Added plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,085] INFO Loading plugin from: /Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/debezium-connector-mongodb-1.0.0.Final.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:241)
[2020-01-10 08:14:08,120] INFO Registered loader: PluginClassLoaderpluginLocation=file:/Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/debezium-connector-mongodb-1.0.0.Final.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:264)
[2020-01-10 08:14:08,121] INFO Added plugin 'io.debezium.connector.mongodb.MongoDbConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,121] INFO Added plugin 'io.debezium.connector.mongodb.transforms.ExtractNewDocumentState' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,121] INFO Added plugin 'io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,122] INFO Loading plugin from: /Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/debezium-core-1.0.0.Final.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:241)
[2020-01-10 08:14:08,198] INFO Registered loader: PluginClassLoaderpluginLocation=file:/Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/debezium-core-1.0.0.Final.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:264)
[2020-01-10 08:14:08,198] INFO Added plugin 'io.debezium.converters.ByteBufferConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,199] INFO Added plugin 'io.debezium.transforms.UnwrapFromEnvelope' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,199] INFO Added plugin 'io.debezium.transforms.ExtractNewRecordState' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,199] INFO Added plugin 'io.debezium.transforms.outbox.EventRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,199] INFO Added plugin 'io.debezium.transforms.ByLogicalTableRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,200] INFO Loading plugin from: /Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/mongo-kafka-0.2-all.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:241)
[2020-01-10 08:14:08,340] INFO Registered loader: PluginClassLoaderpluginLocation=file:/Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/mongo-kafka-0.2-all.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:264)
[2020-01-10 08:14:08,340] INFO Added plugin 'com.mongodb.kafka.connect.MongoSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,340] INFO Added plugin 'com.mongodb.kafka.connect.MongoSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:193)
[2020-01-10 08:14:08,341] INFO Loading plugin from: /Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/bson-3.11.1.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:241)
[2020-01-10 08:14:08,373] INFO Registered loader: PluginClassLoaderpluginLocation=file:/Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/bson-3.11.1.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:264)
[2020-01-10 08:14:08,373] INFO Loading plugin from: /Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/mongodb-driver-core-3.11.1.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:241)
[2020-01-10 08:14:08,465] INFO Registered loader: PluginClassLoaderpluginLocation=file:/Users/1111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0/mongodb-driver-core-3.11.1.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:264)

但最后还是同样的问题

    [2020-01-10 08:40:43,613] ERROR Plugin class loader for connector: 'io.debezium.connector.mongodb.MongoDbConnector' was not found. Returning: org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader@33f2df51 (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:165)
[2020-01-10 08:40:43,809] ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:121)
java.lang.IllegalArgumentException: username can not be null
    at com.mongodb.MongoCredential.<init>(MongoCredential.java:350)
    at com.mongodb.MongoCredential.<init>(MongoCredential.java:344)
    at com.mongodb.MongoCredential.createCredential(MongoCredential.java:169)
    at io.debezium.connector.mongodb.ConnectionContext.<init>(ConnectionContext.java:69)
    at io.debezium.connector.mongodb.MongoDbConnector.validate(MongoDbConnector.java:222)
    at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:313)
    at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig(StandaloneHerder.java:192)
    at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:115)
[2020-01-10 08:40:43,810] INFO Kafka Connect stopping (org.apache.kafka.connect.runtime.Connect:66)

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

["class":"com.mongodb.kafka.connect.MongoSinkConnector","type":"sink","version":"0.2","class":"com.mongodb.kafka.connect.MongoSourceConnector","type":"source","version":"0.2","class":"io.confluent.connect.activemq.ActiveMQSourceConnector","type":"source","version":"5.3.2","class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","type":"sink","version":"5.3.2","class":"io.confluent.connect.ibm.mq.IbmMQSourceConnector","type":"source","version":"5.3.2","class":"io.confluent.connect.jdbc.JdbcSinkConnector","type":"sink","version":"5.3.2","class":"io.confluent.connect.jdbc.JdbcSourceConnector","type":"source","version":"5.3.2","class":"io.confluent.connect.jms.JmsSourceConnector","type":"source","version":"5.3.2","class":"io.confluent.connect.replicator.ReplicatorSourceConnector","type":"source","version":"5.3.2","class":"io.confluent.connect.s3.S3SinkConnector","type":"sink","version":"5.3.2","class":"io.confluent.connect.storage.tools.SchemaSourceConnector","type":"source","version":"5.3.2-ce","class":"io.debezium.connector.mongodb.MongoDbConnector","type":"source","version":"1.0.0.Final","class":"org.apache.kafka.connect.file.FileStr

提前致谢

【参考方案1】：

首先，请使用 Kafka Connect REST 接口检查插件的安装情况（详情请参阅here）。

尝试使用plugin path 机制而不是CLASSPATH 安装Kafka Connect 插件（docs 中的更多信息）。

【讨论】：

感谢您的回答。我也尝试过使用插件路径，下面是connect-standalone.properties和connect-avro-standalone.properties中的路径，仍然是同样的问题，它无法为连接器加载插件类加载器：'io.debezium.connector.mongodb.MongoDbConnector... 虽然我看到它能够在日志开头从指定位置注册所有必需的加载器并添加插件。 plugin.path=/Users/11111111/workspace/KafkaConnect/confluent-5.3.2/debezium-debezium-connector-mongodb-1.0.0

试试plugin.path=/Users/11111111/workspace/KafkaConnect/confluent-5.3.2。你的插件应该在plugin.path 里面。例如。如果你有两个插件和plugin.path=/opt/connectors，第一个插件的路径是/opt/connectors/confluentinc-kafka-connect-http-1.0.3/，第二个路径是/opt/connectors/confluentinc-kafka-connect-jdbc-5.3.1/

@ Iskuskov Alexander - 你是对的，当我更改路径时，它会尝试从 confluent- 5.3.2/... 中的每个目录加载插件但是在这两种情况下我都看到它能够找到并加载来自 debezium 的插件，我在原始问题中添加了详细的日志.. 仍然是同样的问题。

@user837593，请显示curl localhost:8083/connector-plugins 命令的输出。

@ Iskuskov Alexander -Kafka 连接在连接器错误后停止，因此端口 8083 不可用 [2020-01-10 08:40:43,810] 信息 Kafka 连接停止（org.apache.kafka.connect.runtime .Connect:66) [2020-01-10 08:40:43,810] 信息停止 REST 服务器 (org.apache.kafka.connect.runtime.rest.RestServer:244)

【参考方案2】：

有两个问题

Debezium 连接器已硬编码 source='admin'，无法覆盖它，而且我尝试连接的数据库服务器没有任何管理员身份验证。

我使用的测试服务器是独立服务器并且没有副本，因此我必须创建副本并将此服务器添加到其中。 Debezium 连接器不支持独立服务器。

【讨论】：

第一个（可配置的身份验证源）已为即将发布的 Debezium 1.1.0.Alpha1 版本修复。