使用Kafka主题的所有消息并断开连接

Question

我有一个批处理作业，它将一天触发一次。要求是

1. 使用该时间点上关于Kafka主题的所有可用消息
2. 处理消息
3. 如果进程已成功完成，则提交偏移量。

当前，我poll（）while循环中的消息，直到ConsumerRecords.isEmpty（）为true。当ConsumerRecords.isEmpty（）为true时，我假设Topic在该时间点的所有可用记录都已被使用。应用程序维护偏移量并关闭kafka使用者。

当消息处理完成并成功完成时，我创建一个新的KafkaConsumer并提交应用程序维护的偏移量。

注意，我关闭了最初用于读取消息的KafkaConsumer，并使用另一个KafkaConsumer实例提交偏移，以避免消费者重新平衡异常。

我期待最多5K消息的主题。主题被分区和复制。

有没有更好的方法在特定的时间点消耗所有关于主题的消息？有没有我遗漏的或者需要照顾的东西？我认为我不需要处理消费者的再平衡，因为我为循环中的消息poll（）并在轮询完成后处理消息。

我使用的是JavaKafka客户端V0.9，如果在上面的场景中有帮助的话，可以改成V0.10。

谢谢

更新：

AtomicBoolean flag = new AtomicBoolean();
flag.set(true);

while(flag.get()) {

 ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(timeout);

 if(consumerRecords.isEmpty()) {
     flag.set(false);
     continue;
 }
  //if the ConsumerRecords is not empty process the messages and continue to poll()
}
kafkaConsumer.close();

Answer 1

您不能假设在调用poll（）之后，由于使用者上的max.poll.records配置参数，您已经读取了主题中的所有可用消息。这是单个poll（）返回的最大记录数，其默认值为500。这意味着，如果在这个时刻主题中有600条消息，您需要对poll（）进行两次调用以读取所有消息（但同时考虑到可能会有其他消息到达）。我不明白的另一件事是为什么要使用不同的消费者来提交偏移。你说的消费者再平衡例外是什么？