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

从Apache Nifi中上次提交的偏移量读取consumer中的Kafka消息，可以通过以下步骤实现：

1. 首先，确保已经安装和配置了Apache Nifi和Kafka，并且它们能够正常运行。
2. 在Apache Nifi中创建一个Kafka Consumer组件，用于从Kafka主题中读取消息。配置Kafka Consumer的相关属性，包括Kafka集群的地址、主题名称、消费者组ID等。
3. 在Kafka Consumer组件的属性中，找到"Offset Reset"选项。将其设置为"Latest"，这样当没有提交的偏移量时，将从最新的消息开始读取。
4. 在Kafka Consumer组件的属性中，找到"Auto Commit"选项。将其设置为"false"，这样可以手动控制偏移量的提交。
5. 在Nifi中创建一个自定义属性，例如"offset"，用于存储偏移量的值。
6. 在Kafka Consumer组件的属性中，找到"Initial Offset"选项。将其设置为${offset}，这样可以从存储的偏移量值开始读取消息。
7. 在Kafka Consumer组件的输出端口上添加一个UpdateAttribute处理器，用于更新偏移量的值。
8. 在UpdateAttribute处理器的属性中，设置一个新的属性，例如"new_offset"，将其值设置为${kafka.offset}，这样可以获取到当前读取的偏移量。
9. 在UpdateAttribute处理器的属性中，设置一个新的属性，例如"offset"，将其值设置为${new_offset}，这样可以更新存储的偏移量值。
10. 在Kafka Consumer组件的成功输出端口上添加一个PutKafka处理器，用于将读取的消息发送到另一个Kafka主题。

通过以上步骤，可以实现从Apache Nifi中上次提交的偏移量读取consumer中的Kafka消息。在每次处理完一批消息后，手动提交偏移量，以确保下次读取消息时能够从正确的位置开始。