监控Kafka主题的消费者数量

Question

我们正在使用 Prometheus 和 Grafana 来监控我们的 Kafka 集群。

在我们的应用程序中，我们使用 Kafka 流，Kafka 流有可能由于异常而停止。我们正在记录事件setUnCaughtExceptionHandler，但是，当流停止时，我们还需要某种警报。

我们目前拥有的是，jmx_exporter 作为代理运行并通过端点公开 Kafka 指标，prometheus 从端点获取指标。

我们没有看到任何衡量每个主题的活跃消费者数量的指标。我们错过了什么吗？有关如何获取活跃消费者数量并在消费者停止时发送警报的任何建议。

Answer 1

我们有类似的需求，并将每个分区的 Kafka Consumer Lag 添加到 Grafana 中，如果延迟超过指定阈值，还会添加警报（每个主题的阈值应该不同，具体取决于负载，例如，对于某些主题，它可能是 10，对于高负载 - 100000）。所以如果你有更多，例如1000 条未处理的消息，您将收到警报。

您可以为每个 kafka 流添加状态侦听器，如果流处于错误状态，记录错误或发送电子邮件：

kafkaStream.setStateListener((newState, oldState) -> {
    log.info("Kafka stream state changed [{}] >>>>> [{}]", oldState, newState);
    if (newState == KafkaStreams.State.ERROR || newState == KafkaStreams.State.PENDING_SHUTDOWN) {
        log.error("Kafka Stream is in [{}] state. Application should be restarted", newState);
    }
});

您还可以添加运行状况检查指示器（例如，通过 REST 端点或通过 spring-boot HealthIndicator），以提供流是否正在运行的信息：

KafkaStreams.State streamState = kafkaStream.state(); state.isRunning();

我还没有找到任何 kafka 流指标来提供有关活动消费者或可用连接分区的信息，但对我而言，如果 kafka 流提供此类数据会很好（并希望它在未来的版本中可用）。