更新FlowFile属性存入Kafka后，ConsumeKafka_2.0消费时属性丢失咨询

核心原因：PutKafka默认不会传递FlowFile属性到Kafka记录 #

你遇到的问题并不是ConsumeKafka_2.0不支持属性读取，而是你在使用PutKafka（或类似的Kafka写入处理器）时，没有配置将FlowFile的自定义属性写入到Kafka记录中。ConsumeKafka_2.0的getAttributes(record)方法确实会从Kafka记录里提取属性，但前提是这些属性已经被写入到Kafka的记录结构（比如Headers）中了——如果Put的时候没传，消费时自然读不到。

解决步骤：配置PutKafka传递属性 #

要让FlowFile属性随Kafka记录一起传递，你需要在写入Kafka的处理器（比如PutKafka_2.0）中做以下配置：

• 设置Attribute Encoding：选择属性的编码方式（比如UTF-8）
• 配置Attributes to Include：指定要传递的自定义属性（可以用*表示所有属性，或者逗号分隔特定属性名）
• 如果是通过Kafka Headers传递属性，设置Attribute Header Prefix：比如nifi.attr.，这样处理器会把FlowFile属性以nifi.attr.属性名的形式写入Kafka记录的Headers中

对应配置ConsumeKafka_2.0读取属性 #

在ConsumeKafka_2.0中，需要和Put端的配置对应：

• 设置相同的Attribute Encoding
• 设置相同的Attribute Header Prefix：这样处理器会识别Kafka Headers中带有该前缀的键，去掉前缀后作为FlowFile的属性
• 确保Extract Attributes选项处于启用状态（默认应该是启用的）

针对你贴的源码解释 #

你看到的session.putAllAttributes(flowFile, getAttributes(record))代码确实是在把从Kafka记录中提取的属性写入FlowFile，但getAttributes(record)的返回值完全依赖于Kafka记录本身是否包含这些属性。如果PutKafka没有把自定义属性写入Kafka记录，这个方法只会返回一些默认属性（比如kafka.topic、kafka.partition等），而你的自定义属性自然不会出现在FlowFile中。

是否需要自定义处理器？ #

不需要！只要正确配置Put和Consume两端的Kafka处理器，就能实现FlowFile属性的传递。只有当你需要非常特殊的属性存储格式（比如把属性编码到记录内容的特定位置）时，才需要考虑自定义处理器。

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