NiFi ConsumeKafka_2_6处理器无法正确提取Long类型记录头

问题根源 #

Kafka中Scala Long类型的头是二进制序列化存储的，而NiFi的「Headers to Add as Attributes (Regex)」属性默认会直接将二进制内容按ASCII编码转成字符串，导致出现乱码字符，无法直接还原为原始Long值。

可行解决办法 #

• 方案一：生产者端提前转换为字符串
  在Scala生产者代码中，将Long类型的头值直接转为字符串后再发送（例如yourLongValue.toString）。这样NiFi提取属性时直接拿到可读的数字字符串，无需后续转换。

• 方案二：NiFi端二进制头解析流程

  1. 调整ConsumeKafka_2_6配置：移除「Headers to Add as Attributes (Regex)」的.*配置，或仅匹配字符串类型的头名称，让Kafka头以二进制形式保留在FlowFile的kafka.headers属性中。
  2. 添加ExtractKafkaHeaders处理器，指定要提取的目标Long类型头名称，提取后会生成类似kafka.header.xxx的属性（二进制格式）。
  3. 使用ExecuteScript处理器（推荐Groovy/Scala脚本）解析二进制属性：

     def targetHeader = "kafka.header.your_long_header_name"
     def headerValue = flowFile.getAttribute(targetHeader)
     if (headerValue != null) {
         byte[] bytes = headerValue.getBytes("ISO-8859-1")
         long parsedValue = ByteBuffer.wrap(bytes).getLong()
         flowFile = session.putAttribute(flowFile, "parsed_long_value", String.valueOf(parsedValue))
     }
     return flowFile
     

  4. 后续流程直接使用解析后的parsed_long_value属性即可。

• 方案三：使用表达式语言直接转换
  通过ExtractKafkaHeaders提取二进制头属性后，添加ConvertAttribute处理器，用NiFi表达式语言完成转换：
  目标属性填写parsed_long_value，值填写${kafka.header.your_long_header_name:toBytes():toLong()}，即可直接将二进制字符串转为Long值并存储为新属性。

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