在 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 或成员。