我尝试使用ConsumeKafka_1_0、ConsumeKafka_2_0和ConsumeKafka_2_6的版本
取决于您使用的Kafka版本。如果您使用的Kafka broker版本高于2.6，则使用该版本。如果您的版本高于2.0，但低于2.6，则使用2.0......依此类推。
将“偏移重置”设置为“最新”或“最早”
取决于您是否关心现有数据。如果是，请使用earliest。
修改了每个请求上的组ID
这是一个很好的调试步骤，但不是所有时候都需要。如果组实际上是在Kafka群集中创建的，您可以使用kafka-consumer-groups.sh命令进一步调试。
将事务处理设为True或False
取决于您的生成器是否使用事务。如果您将此设置为True，而您的生成器不使用它们，则不确定NiFi将执行什么操作...不过，将此设置保留为True可能是安全的。

包含最新Kafka作品的示例

对于纯文本Kafka协议

• 安全.协议=明文
• sasl.mechanism =PLAIN（无法禁用，因此需要用户名+密码）
• 用户名：某个非空字符串
• 密码：某个非空字符串

用于简单的Kafka消费者调试

• 记录读取器：GrokReader
• 使用Grok表达式进行配置：%{GREEDYDATA:message}

对于内部Nifi FlowFile输出（可以使用任何格式，但JSON适合验证您的步骤是否有效）

• 记录作者：JSONRecordSetWriter

为parse.failure添加新的处理器（或在“关系”选项卡下终止它），并通过单击并从“使用处理器”中拖动来设置success关系。您可以禁用下一个处理器，以便在NiFi中简单地将FlowFiles排队。
然后，启动Consume处理器，并检查队列中的数据。
生产者命令：

echo 'Hello, World' | kcat -P -b localhost:9092 -t foobar