使用Kafka-MongoDB-Connect向MongoDB插数据遇错误求助

我来帮你梳理下当前的配置和数据情况，然后拆解可能导致插入失败的核心问题：

一、当前环境配置详情 #

1. Kafka Connect 全局配置 #

group.id=stage
offset.storage.topic=stage_offset
config.storage.topic=stage_config
status.storage.topic=stage_status
bootstrap.servers=****************:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
rest.host.name=localhost
rest.port=9002
plugin.path=/mnt/d/confluent-5.2.1/plugins
offset.storage.file.filename=/tmp/connect.offsets

2. MongoDB Sink Connector 配置 #

{
  "name": "mongo-sink-connector",
  "config": {
    "connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
    "tasks.max": "1",
    "topics": "VENTE_PDV_JOUR",
    "mongodb.connection.uri": "mongodb://localhost:27018/stage-project?w=1&journal=true",
    "mongodb.value.projection.type": "whitelist",
    "mongodb.value.projection.list":"STOREID,BUSINESSDATE,CA",
    "mongodb.collection": "ticket-data",
    "flush.size": "100",
    "rotate.interval.ms": "1000",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable": false,
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter.schemas.enable": false
  }
}

二、Kafka主题数据情况 #

原始分隔符格式数据 #

1557408134887 | 7316|+|2019-05-09 | 7316 | 2019-05-09 | 249.35999999999999
1557408134941 | 6110|+|2019-05-09 | 6110 | 2019-05-09 | 3321.0399999999995
1557408135199 | 11688|+|2019-05-09 | 11688 | 2019-05-09 | 3261.800000000001
1557408134603 | 55247|+|2019-05-09 | 55247 | 2019-05-09 |995.4200000000002

期望插入的JSON格式数据 #

{"STOREID":3194,"BUSINESSDATE":"2019-05-09","CA":"1.82"}
{"STOREID":4479,"BUSINESSDATE":"2019-05-09","CA":"7.71"}
{"STOREID":6818,"BUSINESSDATE":"2019-05-09","CA":"121.69"}
{"STOREID":9127,"BUSINESSDATE":"2019-05-09","CA":"67.82"}
{"STOREID":4987,"BUSINESSDATE":"2019-05-09","CA":"40.56"}
{"STOREID":1654,"BUSINESSDATE":"2019-05-09","CA":"26.94"}

三、核心问题排查方向 #

1. 数据格式不匹配（最可能的原因） #

你当前Kafka主题里的原始数据是自定义分隔符的文本格式，但MongoDB连接器配置的value.converter是StringConverter——它只会把数据当作纯字符串传递给连接器，而连接器需要识别STOREID这类JSON字段才能完成投影和插入。同时全局配置的JsonConverter和连接器局部的StringConverter存在逻辑冲突，进一步加剧了解析问题。

快速验证方案：
手动往主题发送一条标准JSON数据，测试连接器是否能正常插入：

kafka-console-producer --broker-list ****************:9092 --topic VENTE_PDV_JOUR

输入：

{"STOREID":3194,"BUSINESSDATE":"2019-05-09","CA":"1.82"}

如果这条数据能成功写入MongoDB，说明连接器本身正常，问题出在原始数据的格式转换上。

格式转换解决建议：

• 使用KSQL将分隔符数据转换为JSON流：

CREATE STREAM VENTE_PDV_JOUR_JSON AS
SELECT 
    TRIM(SPLIT(value, '|')[1]) AS STOREID,
    TRIM(SPLIT(value, '|')[3]) AS BUSINESSDATE,
    TRIM(SPLIT(value, '|')[6]) AS CA
FROM VENTE_PDV_JOUR;

然后让MongoDB连接器消费这个新的VENTE_PDV_JOUR_JSON主题。

• 或者添加Kafka Connect Transform，比如RegexExtract配合ReplaceField来提取并映射字段。

2. MongoDB连接有效性验证 #

确认MongoDB实例确实运行在localhost:27018，stage-project数据库已存在，且连接器拥有ticket-data集合的读写权限。可以用客户端手动测试连接：

mongo mongodb://localhost:27018/stage-project

如果连接失败，检查MongoDB的监听地址、端口和防火墙设置。

3. 插件兼容性检查 #

你使用的at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector是社区版插件，需要确认它的版本与Confluent 5.2.1兼容。版本不匹配可能导致类加载错误或运行时异常。

4. Kafka主题配置检查 #

确认stage_offset、stage_config、stage_status这三个内部主题已创建，且分区/副本数符合要求（建议每个主题至少3个副本）。可以用以下命令查看：

kafka-topics --list --bootstrap-server ****************:9092

如果主题不存在，需确保Kafka集群的auto.create.topics.enable设置为true。

5. 日志细节排查 #

直接查看Kafka Connect的日志文件（默认在Confluent的logs目录下，文件名为connect.log或connect-distributed.log），日志会详细记录错误原因——比如连接失败、数据解析错误、权限不足等，是定位问题的最直接方式。

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