MongoSourceConnector墓碑消息失效：删除文档未发布Null值问题

核心问题分析 #

当你设置publish.full.document.only=true和publish.full.document.only.tombstone.on.delete=true时无法生成Null墓碑消息，主要和以下配置冲突或限制有关：

• 自定义Schema与null值的兼容性问题
• Transform对null消息的处理逻辑限制
• 输出格式与转换器的Schema配置不匹配

具体排查与解决步骤 #

1. 检查Transform对Null消息的影响 #

你配置的RenameFieldValueID是针对Value的字段替换操作，但墓碑消息的Value为null，该Transform无法处理null值，可能直接丢弃了消息。

验证方法：临时移除transforms.RenameFieldValueID相关配置，重启连接器后测试删除操作，若能生成墓碑消息，说明该Transform是阻碍因素。

2. 调整输出格式与Schema配置 #

你当前使用output.format.value=schema并自定义了严格的Record Schema，但墓碑消息的Value是null，而你的Schema中未允许整个Record为null，连接器无法将null值匹配到该Schema，导致消息无法生成。

推荐调整：将output.format.value改为json，同时保留publish.full.document.only=true和publish.full.document.only.tombstone.on.delete=true。JSON格式对null值兼容性更好，连接器会直接生成Key为文档ID、Value为null的墓碑消息。

3. 修正转换器的Schema开关 #

你设置了value.converter.schemas.enable=false，但搭配output.format.value=schema时会产生冲突——连接器输出带Schema的消息，而转换器关闭了Schema支持，导致null值无法正确序列化。改用output.format.value=json后，该配置可正常工作。

修改后的示例配置 #

{
  "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
  "publish.full.document.only": "true",
  "publish.full.document.only.tombstone.on.delete": "true",
  "output.format.value": "json",
  "tasks.max": "1",
  "output.format.key": "schema",
  "output.schema.key": "{\"type\":\"record\",\"name\":\"keySchema\",\"fields\":[{\"name\":\"documentKey._id\",\"type\":\"string\"}]}",
  "key.converter.schemas.enable": "false",
  "transforms": "ExtractFieldKeyID",
  "transforms.ExtractFieldKeyID.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.ExtractFieldKeyID.field": "documentKey._id",
  "database": "ais-user",
  "output.json.formatter": "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",
  "connection.uri": "",
  "value.converter.schemas.enable": "false",
  "name": "ais-user.administrator",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter"
}

补充说明 #

若必须保留RenameFieldValueID Transform，可添加过滤逻辑，让null消息跳过该操作：

"transforms": "ExtractFieldKeyID,FilterNull,RenameFieldValueID",
"transforms.FilterNull.type": "org.apache.kafka.connect.transforms.Filter$Value",
"transforms.FilterNull.condition": "$value != null"

这样只有非null消息会进入字段重命名流程，null墓碑消息可直接通过。

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