如何在NiFi中基于Kafka墓碑消息（Null消息）实现路由？

Kafka墓碑消息指的是消息体为Null的记录，在NiFi里可以通过以下方式实现识别和路由：

1. 利用ConsumeKafka内置关联关系（最简方案） #

配置ConsumeKafka（或对应版本如ConsumeKafka_2_0）的Null Value Behavior属性为Route to 'null' relationship。此时处理器会自动将接收到的Null消息（墓碑消息）路由到null分支，非Null消息则走success分支，直接通过连接这两个分支到后续组件即可完成路由。

2. 自定义判断逻辑（灵活扩展） #

如果需要更复杂的校验逻辑（比如同时判断消息键或附加属性），可以搭配RouteOnAttribute处理器：

• 添加路由规则，命名为is_tombstone
• 规则表达式使用${content.size():equals(0)}——NiFi中Kafka墓碑消息对应的FlowFile内容大小为0
• 将匹配规则的FlowFile路由到指定分支（如Tombstone），不匹配的走unmatched分支

也可以用ExecuteScript处理器编写Groovy脚本实现自定义判断：

def flowFile = session.get()
if (!flowFile) return

def contentStream = session.read(flowFile)
def isTombstone = contentStream.available() == 0

if (isTombstone) {
    session.transfer(flowFile, REL_SUCCESS) // 路由到墓碑消息处理分支
} else {
    session.transfer(flowFile, REL_FAILURE) // 路由到正常消息处理分支
}

注意事项 #

• 不同版本的ConsumeKafka处理器属性名称可能略有差异，比如旧版本可能叫Null Message Handling，需对应调整
• 上述逻辑仅判断消息体为Null的场景，若需识别键为Null但值不为Null的情况，需额外添加对kafka.key属性的判断

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