Kubernetes中Kafka Connect配置Mongo Sink Connector报topic未定义错误求助

看起来你的配置存在两个关键问题，我们逐个拆解排查：

1. "Configuration is not defined: topic" 错误提示 #

你当前配置中使用的是复数形式的topics参数，但错误提示要求单数的topic——这大概率是版本兼容性或参数名适配问题：

• 先确认Kafka Connect版本与Mongo Kafka Connector 1.1.0的兼容性：Mongo 1.1.0 Connector要求Kafka Connect版本在2.3.x到2.5.x之间，如果你的Kafka Connect版本超出这个范围，可能会出现参数不匹配的情况。
• 尝试将配置里的"topics":"<topic name>"替换为"topic":"<topic name>"（单数形式），部分早期版本的Mongo Sink Connector确实使用单数参数名。

2. "Unable to connect to the server" 错误 #

这个问题的核心是MongoDB连接地址的配置不符合Kafka Connect的运行环境要求：

• Kubernetes集群内的访问地址：你的Kafka Connect部署在Kubernetes集群中，所以connection.uri里的MongoDB地址必须是Kubernetes集群能够访问的地址（比如MongoDB的Service名称、集群外部可访问的公网地址），而不是你的本地地址——虽然你通过端口转发在本地发起curl请求，但Connector的运行环境是K8s集群，它需要自己能直接连接到MongoDB。
• 特殊字符URL编码：如果你的MongoDB用户名或密码包含@、:、/这类特殊字符，必须进行URL编码（比如@替换为%40，:替换为%3A），否则会导致连接字符串解析失败。
• 权限验证：确保你使用的MongoDB用户拥有目标数据库和集合的写入权限，权限不足也会触发连接失败的提示。

修正后的示例配置 #

这里提供一个调整后的配置参考，你可以根据实际情况修改：

{ 
  "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector", 
  "tasks.max":"1", 
  "topic":"<topic name>",  // 根据版本调整为单数或复数
  "connection.uri":"mongodb://<编码后的用户名>:<编码后的密码>@<K8s可访问的Mongo地址>:<port>", 
  "database":"<database name>", 
  "collection":"<collection name>", 
  "key.converter":"org.apache.kafka.connect.storage.StringConverter", 
  "key.converter.schemas.enable":false, 
  "value.converter":"org.apache.kafka.connect.json.JsonConverter", 
  "value.converter.schemas.enable":false 
}

另外，你可以使用Kafka Connect的配置验证端点提前检查配置有效性，命令如下：

curl -X POST http://localhost:8083/connectors/MongoSinkConnector/config/validate -H "Content-Type: application/json" -d '<你的配置JSON>'

这个命令会返回更详细的错误细节，帮助你快速定位问题。

内容的提问来源于stack exchange，提问作者David