Flink-Query Kafka主题用于消费者群体的偏移？

Question

我有一个用例，其中数据将从kafkaTopic1流入程序（我们称之为P1)，经过处理，然后持久化到数据库。P1将在一个多节点集群上，因此每个节点将处理大量的kafka分区（假设本主题有5个节点和50个kafka分区）。如果其中一个节点由于任何原因完全失败，并且有数据正在处理，那么该数据将丢失。

例如，如果kafkaTopic1上有500条消息，node2拉出了10条消息（因此根据偏移量要拉出的下一条消息是Message11)，但只有其中8条被完全处理并在节点失败时持久保存到数据库，则仍在处理的2条将丢失。当节点被恢复时，它将开始从消息11读取，跳过丢失的两条消息（从技术上讲，kafka分区将开始将其消息发送到另一个要处理的节点，因此该分区的偏移量将移动，我们不一定知道节点死亡时下一个要处理的消息是什么）。

（注意：当节点死亡时，假设用户注意到并完全关闭P1，因此此时暂时不会处理更多数据）。

Answer 1

如果在flink作业中启用了检查点，那么您不应该丢失消息，因为flink也在内部维护偏移量，并且从失败中恢复后，它应该读取flink最后提交的偏移量。

现在，如果您仍然希望找到偏移量并重新开始从偏移量中读取，这将变得很棘手，因为您需要根据给定的使用者组为给定的主题找到所有分区的偏移量。

我不知道如何从开箱即用的Flink-kafka-Consumer API中做到这一点，但您可以将kafka依赖项添加到项目中，并从kafka API创建一个kafkaconsumer。一旦有了消费者，就可以调用

consumer.position(partition) 

consumer.committed(partition)

请注意，您仍然需要遍历所有分区以获得所有当前偏移量

在这里阅读差异：Kafka Javadoc

一旦有了要从中读取的偏移量，就可以在flink作业中手动指定使用者偏移量，方法如下：

Map<KafkaTopicPartition, Long> specificStartOffsets = new HashMap<>();
specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L);
specificStartOffsets.put(new KafkaTopicPartition("myTopic", 1), 31L);
specificStartOffsets.put(new KafkaTopicPartition("myTopic", 2), 43L);

myConsumer.setStartFromSpecificOffsets(specificStartOffsets);