执行Kafka消费者健康检查的最佳方式。

Question

假设我有一个使用来自kafka集群的日志的应用程序。我希望应用程序定期检查群集的可用性，并根据它执行某些操作。我想到了一些方法，但不确定哪一个更好或者最好的方法是什么：

1. 创建MessageProducer和MessageConsumer。生产者将heartbeatTopic发布到集群，消费者会查找它。我认为这个问题是，应用程序只涉及消费，健康检查既有生产也有消费。
2. 使用新的groupId创建一个MessageConsumer，该groupId不断地为新消息进行池化。这样监视/健康检查就像应用程序应该做的那样，我认为这是好的。
3. 创建一个MessageConsumer，它执行与实际使用消息不同的操作。像listTopics（https://stackoverflow.com/a/47477448/2094963）之类的东西。

哪种方法更可取，为什么？

Answer 1

在这里稍微不同的路线，你可以使用Apache Curator library轮询zookeeper（znode路径 - /经纪人/ ids）获取此信息。

这是我尝试和工作的想法 - 我使用Curator's Leader Latch recipe来达到类似的要求。

您可以创建LeaderLatch的实例并调用getLeader()方法。如果在每次调用时，您都会获得一个领导者，则可以安全地假设群集已启动并正在运行，否则会出现问题。

我希望这有帮助。

编辑：添加存储领导者信息的zookeeper节点路径。