检测 Kafka 集群已关闭的最佳方法是什么？

Question

Kafka 消费者 API 非常好，可以隐藏任何暂时的连接错误，并且如果 Kafka 代理死亡并再次出现，只需从其当前偏移量中读取数据。

但在某些应用程序中，如果整个 Kafka 集群已关闭（即所有代理），则发出警报并停止处理数据（来自其他来源）很重要。 我浏览了杂项。 API，这似乎不是一项功能。

我最接近的方法是提交管理员调用，并根据超时得出 Kafka 集群已关闭的结论：

Properties properties   = ... // Load properties from somewhere.
int timeout             = 5_000; // 5 second timeout
AdminClient adminClient = AdminClient.create(properties);
try {
    adminClient.listTopics(new ListTopicsOptions().timeoutMs(timeout)).listings().get();
    // Here we know the cluster is up as call returned within timeout.
} catch (ExecutionException ex) {
    // Here we know that the cluster is down as the call timed out.
}

这是最好的方法吗？

另一种方法是查询 ZooKeeper，但上述方法也适用于应用程序和 Kafka 之间存在网络问题的情况。

Answer 1

虽然我建议您使用适当的监控工具，但如果您仍想以编程方式执行此操作，一种选择是使用 AdminClient 并尝试获取主题名称。

---

例如，

Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
properties.put("request.timeout.ms", 5000);

try {

    AdminClient adminClient = AdminClient.create(properties)

    ListTopicsResult topics = adminClient.listTopics();
    Set<String> names = topics.names().get();

} catch(InterruptedException | ExecutionException e) {
    System.err.println("Kafka is unavailable");
}

但是请注意，如果某些代理关闭，上述内容不会引发异常（但显然，如果代理关闭并不意味着 Kafka 集群本身已关闭，因为数据应该仍然可以访问）

Answer 2

您的方法看起来不错。类似的方法（使用 Spring 的 HealthIndicator 的概念）是 MartinX3 所做的 here。

他的解决方案：

@Component
public class KafkaHealthIndicator implements HealthIndicator {
    private final Logger log = LoggerFactory.getLogger(KafkaHealthIndicator.class);

    private KafkaTemplate<String, String> kafka;

    public KafkaHealthIndicator(KafkaTemplate<String, String> kafka) {
        this.kafka = kafka;
    }

    /**
     * Return an indication of health.
     *
     * @return the health for
     */
    @Override
    public Health health() {
        try {
            kafka.send("kafka-health-indicator", "❥").get(100, TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            return Health.down(e).build();
        }
        return Health.up().build();
    }
}

您可能还希望在返回 Health.up().build()（例如 ActiveControllerCount = 0）之前结合其他 metric-checks，具体取决于您认为对您的用例而言重要的内容以将整个集群视为已关闭.