DocumentDB 5.0.0集成Kafka连接器遇$changeStream.fullDocumentBeforeChange未知字段错误

问题核心 #

你遇到的报错根源是Amazon DocumentDB 5.0.0并未支持MongoDB 5.0引入的$changeStream.fullDocumentBeforeChange特性。尽管DocumentDB宣称兼容MongoDB 5.0的大部分功能，但变更流的这个字段不在其支持范围内，而Atlas作为官方MongoDB服务完整实现了该特性，因此相同配置在Atlas上能正常运行。

解决方案 #

1. 移除或禁用相关配置参数
   直接删除连接器配置中的change.stream.full.document.before.change参数，或者将其值设为off，避免连接器向DocumentDB发送不支持的变更流请求。

2. 调整业务逻辑
   如果你的业务依赖变更前的文档数据，需要重新设计逻辑——DocumentDB无法提供这部分内容，只能基于change.stream.full.document参数获取的变更后文档来处理。

修改后的示例配置 #

{
  "name": "example-name",
  "config": {
    "topic.creation.default.partitions": "1",
    "output.data.format": "AVRO",
    "topic.creation.default.replication.factor": "1",
    "value.converter.schema.registry.url": "http://schemaregistry.confluent.svc.cluster.local:8081",
    "key.converter.schema.registry.url": "http://schemaregistry.confluent.svc.cluster.local:8081",
    "name": "example-name",
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "tasks.max": "1",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "transforms": "",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "topic.creation.groups": "example",
    "connection.uri": "***",
    "database": "***",
    "publish.full.document.only": "true",
    "publish.full.document.only.tombstone.on.delete": "true",
    "change.stream.full.document": "whenAvailable",
    "topic.separator": "_",
    "topic.prefix": "",
    "output.format.key": "schema",
    "output.format.value": "schema",
    "output.schema.infer.value": "true",
    "output.json.formatter": "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",
    "mongo.errors.log.enable": "true",
    "heartbeat.topic.name": "_mongo_heartbeat"
  }
}

验证步骤 #

修改配置后重启Kafka连接器，检查日志是否不再出现$changeStream.fullDocumentBeforeChange相关报错，确认数据能正常从DocumentDB同步到Kafka。

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