NiFi 1.21.0使用PublishKafka_2_6复制Kafka墓碑消息异常

问题场景 #

使用NiFi 1.21.0在两个配置了cleanup.policy=compact的Kafka主题间复制数据，发现**墓碑消息（value为null的消息）**被错误地转换为非null值写入目标主题。例如key为666的消息，源主题edi_test_compact中是正常的墓碑消息，但在目标主题edi_test_compact2中变成了非null内容，当前使用PublishKafka_2_6组件完成目标主题的写入操作。

问题原因 #

NiFi的PublishKafka系列组件默认会将空内容的FlowFile处理为空字节数组或默认字符串，而非Kafka标准的墓碑消息（即value=null）。当源主题的墓碑消息被拉取后，NiFi未正确识别并传递null值，反而将空FlowFile转换为非null内容写入目标主题，导致压缩策略失效。

解决方案 #

针对PublishKafka_2_6组件，调整以下关键配置即可解决：

• 将**Null Value Representation**设置为NULL：该配置项决定了空内容FlowFile对应的Kafka消息value类型，选择NULL后，空FlowFile会被直接写入为Kafka墓碑消息（value=null）。
• 确认**Key Attribute**配置准确：如果源消息的key通过NiFi属性传递，需保证该配置的属性名与拉取组件（如ConsumeKafka_2_6）输出的key属性完全一致，确保key能正确传递，让墓碑消息的key与源主题匹配，保障压缩策略正常生效。

验证步骤 #

• 修改PublishKafka_2_6的Null Value Representation参数为NULL
• 重启NiFi数据流，重新复制包含墓碑消息的数据
• 检查目标主题edi_test_compact2，确认key为666的消息已正确转为墓碑消息（value=null）

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