java - 在java代码中检查Kafka消费者组的活跃度

Question

我已经使用执行器服务为 Kafka 消费者编写了一个 Java 代码，该代码将 24/7 运行消耗数据并处理它。

但在某些情况下，例如由于 Kafka 协调器代理节点宕机，作业将失败，即消费者组将死亡，我将无法知道它，并且 java jar 将运行但不执行任何操作。

 INFO [2018-11-07 03:25:58,425] [pool-1-thread-1] (AbstractCoordinator.java:coordinatorDead) - Marking the coordinator broker_host (id: 2147483446 rack: null) dead for group group_name
 INFO [2018-11-07 03:26:10,817] [pool-1-thread-1] (AbstractCoordinator.java:onSuccess) - Discovered coordinator broker_host (id: 2147483446 rack: null) for group group_name.

我想定期检查java代码中的Kafka消费者活跃度。因此，如果 Kafka 消费者组死亡，我将能够使其存活或触发邮件并终止工作。

任何人都可以对此提供任何意见吗？

Answer 1

如果充当消费者组协调员的代理发生故障，该组中的每个消费者都会找到哪个剩余代理将成为其组的新协调员。这种情况对您的团队来说不应该是致命的。

但是，监视消费者组仍然是一种非常常见的事情，以便跟踪滞后、在需要时扩大/缩小或只是确保重要作业实际运行!

最简单的方法是使用 AdminClient API，尤其是:

• describeConsumerGroups() :列出群组的成员和总体状态
• listConsumerGroupOffsets() :这列出了当前的偏移量

或者，您可以运行 kafka-consumer-groups.sh 工具来检索有关组的详细信息。显示当前位置、滞后和客户端 ID 或成员。