如何从 Apache Nifi 中上次提交的偏移量读取消费者中的 Kafka 消息？

Question

我已经开始我的生产者向 Kafka 发送数据，并开始我的消费者来提取相同的数据。当我在 Apache Nifi 中使用 Consumekafka 处理器（kafka 版本 1.0）时，我的脑海中几乎没有与卡夫卡消费者。

Q.1) 当我第一次启动我的 ConsumeKafka 处理器时，如何从开始和当前消息中读取消息？

Q.2) 以及在 Kafka 中消费者关闭的情况下，如何在最后一条消费消息之后读取消息？

我们如何在使用 Apache Nifi 的同时实现以上两个？

Answer 1

ConsumeKafka 处理器有一个名为“Offset Reset”的属性，当消费者组 id 没有先前的偏移量或偏移量不再存在时，将使用该属性。此属性的选项为“Offset Latest”或“Offset Earliest”，默认为最新。

因此，如果您使用以前从未使用过的消费者组 ID 启动 ConsumeKafka 处理器，那么它会从最新消息开始消费。之后，如果您启动和停止处理器，它将从上次消耗的偏移量开始。

如果你想再次使用“偏移重置”来强制它为最早或最新，那么你需要更改消费者组 id，因为否则现有的消费者组将始终使用现有的偏移量开始。

您不能同时从头和当前阅读消息，您可以从头开始并一直阅读到当前，或者从当前开始。这是 Kafka 的工作方式，并不特定于 NiFi。