如果没有所有经纪人,如何关闭kstream

当这种异常发生时,您可以添加关闭挂钩。可能是 Kafka Stream: Graceful shutdown

拜托 Check here

我添加了一个CustomHealthIndicator,它连续地进行连接检查,如果发现无法访问Kafka,它将抛出异常并关闭应用程序。

示例代码如下:

public class CustomHealthIndicator implements HealthIndicator {
    @Override
    public Health health() {
        return Health.up().withDetail("Status Code", "SUCCESS").build();
    }

    private void doHealthCheck() {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);
        AdminClient adminClient =  AdminClient.create(props);
        try {
            DescribeClusterOptions dco = new DescribeClusterOptions();
            dco.timeoutMs(30000);
            adminClient.describeCluster(dco).clusterId().get();
        } catch (Exception e) {
            System.exit(-1);
        } finally {
            adminClient.close();
            adminClient = null;
        }
    }

}