NiFi跨版本Kafka迁移难题：旧版消费与高版本安全生产冲突

方案一：双NiFi集群联动（生产环境优先推荐） #

• 保留NiFi 1.9.2集群，专门处理Kafka 0.8.2.1的消息消费：沿用原有的GetKafka组件配置，确保消息正常拉取
• 通过NiFi的Site-to-Site实现集群间消息转发：
  1. 在NiFi 1.9.2中添加RemoteProcessGroup，配置NiFi 1.18.0集群的Site-to-Site端点地址（如http://nifi118-node:8080/nifi）
  2. 在NiFi 1.18.0中创建InputPort，设置为允许Site-to-Site连接，接收来自1.9.2集群的消息
• 在NiFi 1.18.0中用PublishKafka_2_6组件完成Kafka 3.2.1的消息生产，配置好SASL_SCRAM的用户名、密码及jaas.config参数
• 优势：彻底规避版本兼容问题，各集群职责明确，运维简单
• 劣势：需维护两套NiFi集群，增加少量资源投入

方案二：在NiFi 1.18.0中自定义Kafka 0.8.2.1消费逻辑 #

利用NiFi的ExecuteScript组件，基于Kafka 0.8.x旧版客户端API实现消费：

1. 下载Kafka 0.8.2.1的相关依赖jar包（kafka_2.10-0.8.2.1.jar、scala-library-2.10.4.jar等），放入NiFi 1.18.0的lib目录，重启NiFi
2. 使用Groovy脚本编写消费逻辑（示例）：

import kafka.consumer.ConsumerConfig
import kafka.consumer.ConsumerIterator
import kafka.consumer.KafkaStream
import kafka.javaapi.consumer.ConsumerConnector
import kafka.serializer.StringDecoder
import kafka.utils.VerifiableProperties
import java.util.Properties
import java.util.HashMap

// 配置Kafka 0.8消费参数
def props = new Properties()
props.put("zookeeper.connect", "zk-host:2181") // Kafka 0.8依赖ZooKeeper管理offset
props.put("group.id", "nifi-kafka08-consumer-group")
props.put("auto.offset.reset", "smallest")

def config = new ConsumerConfig(props)
def consumerConnector = kafka.javaapi.consumer.Consumer.createJavaConsumerConnector(config)

// 订阅目标主题
def topicCountMap = new HashMap<String, Integer>()
topicCountMap.put("your-target-topic", 1)

def decoderProps = new VerifiableProperties()
def stringDecoder = new StringDecoder(decoderProps)

// 获取消息流并处理
def streamMap = consumerConnector.createMessageStreams(topicCountMap, stringDecoder, stringDecoder)
def stream = streamMap.get("your-target-topic").get(0)
def it = stream.iterator()

while (it.hasNext()) {
    def messageContent = it.next().message()
    // 创建FlowFile并写入消息内容
    def flowFile = session.create()
    flowFile = session.write(flowFile, { out ->
        out.write(messageContent.getBytes())
    } as OutputStreamCallback)
    session.transfer(flowFile, REL_SUCCESS)
}

// 关闭消费者连接
consumerConnector.shutdown()

3. 配置ExecuteScript组件的调度周期，确保消息持续消费

• 优势：只需维护一套NiFi集群
• 劣势：需自定义脚本并维护依赖版本，问题排查复杂度较高

方案三：尝试在NiFi 1.18.0中引入GetKafka组件（不推荐生产环境） #

把NiFi 1.9.2安装目录下的nifi-kafka-0.8-nar-1.9.2.nar复制到NiFi 1.18.0的lib目录，重启NiFi后查看组件是否可用：

• 风险提示：NiFi 1.9.2和1.18.0的核心依赖、Kafka客户端版本差异极大，大概率会出现依赖冲突导致NiFi启动失败或组件运行异常
• 操作前务必备份NiFi的配置文件和数据目录，准备好回滚方案

内容的提问来源于stack exchange，提问作者chundi sai krishna