Kafka MongoDB Sink Connector数据同步不符合预期问题排查

问题原因 #

• 你看到的带before、after、source的完整结构是Debezium默认输出的变更事件Envelope格式，不是配置错误，默认输出就包含全量变更元数据。
• 删除操作无法识别是因为你没有配置Mongo Sink的删除开关，也没有将MySQL表主键映射为Mongo文档的_id，无法定位待删除的对应文档。
• 日期显示为数字是Debezium默认将MySQL DATE类型转换为了自1970-01-01起的天数数值，没有做格式转换。

修复方案 #

1. 完成Sink连接器剩余配置（解决格式、删除识别问题） #

你已经添加的ExtractNewRecordState转换是对的，作用就是提取Envelope中after字段的实际业务数据，丢弃元数据。补充以下配置即可：

// 在Sink连接器config中新增以下参数
"delete.enabled": "true",
"document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
"document.id.strategy.partial.value.projection.type": "AllowList",
"document.id.strategy.partial.value.projection.list": "id"

参数说明：

• delete.enabled开启后Mongo Sink会识别Debezium的删除事件，自动删除集合中对应的文档
• 后面三个参数是将MySQL表的id主键作为Mongo文档的_id，确保更新、删除操作可以定位到正确的文档，避免重复写入

2. 修复日期格式问题 #

在MySQL源连接器的config中新增以下配置，将日期类型直接序列化为ISO格式字符串或者Mongo支持的日期类型：

// 在源连接器config中新增以下参数
"time.precision.mode": "connect",
"converters": "date",
"date.type": "io.debezium.time.Date",
"date.format.date": "yyyy-MM-dd"

配置完成后重新创建连接器，同步的数据就会直接显示为2001-06-04这类可读的日期格式，也可以根据需要调整为Mongo ISODate类型存储。

最终完整配置参考 #

源连接器配置

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
  "name": "source_mysql_connector",  
  "config": {  
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",  
    "database.hostname": "host.docker.internal",  
    "database.port": "3306",
    "database.user": "test",
    "database.password": "$apr1$o7RbW.GvrPIY1",
    "database.server.id": "8111999",  
    "database.server.name": "db_source",  
    "database.include.list": "example",  
    "database.history.kafka.bootstrap.servers": "broker:29092",  
    "database.history.kafka.topic": "schema-changes.example",
    "database.allowPublicKeyRetrieval":"true",
    "include.schema.changes": "true",
    "time.precision.mode": "connect",
    "converters": "date",
    "date.type": "io.debezium.time.Date",
    "date.format.date": "yyyy-MM-dd"
  }
}'

Sink连接器配置

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
  "name": "sink_mongodb_connector",  
  "config": {  
      "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
      "tasks.max":"1",
      "topics":"db_source.example.employees",
      "connection.uri":"mongodb://172.17.0.1:27017/example?w=1&journal=true",
      "database":"example",
      "collection":"employees",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "value.converter.schema.registry.url": "http://schema-registry:8081",
      "transforms": "unwrap",
      "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
      "transforms.unwrap.drop.tombstones": "false",
      "transforms.unwrap.delete.handling.mode": "rewrite",
      "delete.enabled": "true",
      "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
      "document.id.strategy.partial.value.projection.type": "AllowList",
      "document.id.strategy.partial.value.projection.list": "id"
  }
}'

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