如何让NiFi ConsumeGCPubSub处理器输出eventType属性

针对NiFi 1.25.0版本的ConsumeGCPubSub处理器无法输出GCS PubSub通知中的eventType属性问题，可按以下步骤排查解决：

1. 开启消息属性提取配置 #

ConsumeGCPubSub处理器默认可能未开启提取PubSub消息的属性，需修改以下配置：

• 找到处理器配置中的**Extract Message Attributes**选项，设置为true。
• 检查**Attribute Prefix**配置：如果该字段不为空，提取的属性会带上前缀（比如设为pubsub.，则eventType会变成pubsub.eventType），可根据需求调整前缀或留空直接使用原属性名。

修改完成后重启处理器，查看FlowFile属性中是否出现eventType相关字段。

2. 直接解析消息体获取eventType #

GCS PubSub通知的消息体本身是JSON格式，其中也包含eventType字段。如果上述配置无法生效，可通过后续处理器解析消息体：

• 使用**JoltTransformJSON**处理器，配置Jolt规范提取eventType字段并设置为FlowFile属性。示例Jolt规范：

  {
    "operation": "shift",
    "spec": {
      "eventType": "@(0)"
    }
  }
  

• 或使用**ExecuteScript**处理器（Groovy脚本为例），解析消息体并添加属性：

  import groovy.json.JsonSlurper
  def flowFile = session.get()
  if (!flowFile) return
  flowFile = session.read(flowFile, { inputStream ->
    def json = new JsonSlurper().parse(inputStream)
    flowFile = session.putAttribute(flowFile, "eventType", json.eventType)
  } as InputStreamCallback)
  session.transfer(flowFile, REL_SUCCESS)
  

3. 检查版本兼容性并升级 #

如果以上方法都无效，可能是NiFi 1.25.0版本的ConsumeGCPubSub处理器存在相关bug。可查看NiFi官方release notes，确认后续版本（如1.26.0及以上）是否修复了PubSub属性提取的问题，若有则升级NiFi版本尝试解决。

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