NiFi中如何用属性值动态配置PublishKafka_2_0的Broker与Topic？

针对你遇到的No resolvable bootstrap urls given in bootstrap.servers报错，以下是排查和解决步骤：

• 检查属性引用语法：在PublishKafka_2_0的Bootstrap Servers和Topic Name配置项中，必须用NiFi表达式语言（EL）格式${属性名}来引用流文件属性，比如流文件属性名为kafka_brokers，则配置框里要填${kafka_brokers}，而非直接写属性名。

• 确认流文件属性存在且有效：

  • 确保流文件携带了目标属性，可通过UpdateAttribute处理器预先添加属性（比如设置kafka_brokers值为host1:9092,host2:9092），或用LogAttribute打印所有属性验证是否存在、值是否符合Kafka Broker格式（多个地址用逗号分隔）。
  • 属性名大小写敏感，检查引用的属性名和流文件实际属性名是否完全一致。

• 验证表达式语言（EL）是否启用：PublishKafka_2_0的Bootstrap Servers和Topic Name字段默认支持EL，若曾修改过配置，可检查字段旁的fx图标是否处于激活状态（点击可切换EL启用/禁用）。

• Topic动态设置额外注意：若动态Topic未预先在Kafka创建，需确保Kafka集群配置了auto.create.topics.enable=true，否则会出现Topic不存在的报错。

按以上步骤排查后，重新运行处理器即可实现动态Broker和Topic的配置。

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