java在我的spring boot应用程序中为Kafka实现健康检查的更干净有效的方法是什么？

Question

我有一个SpringBoot（2.1.6）应用程序，它使用（组织范围内的）公共kafka实例并向其生成消息。我正在尝试使用SpringActuator在我的应用程序中为这个kafka代理实现健康检查，我面临着一系列与性能和日志记录相关的问题。SpringBoot2.0内置了一个健康指示器，但由于一些明显的问题，他们将其删除

以下是我实现的healthcheck类：

@Component
public class KafkaHealthCheck implements HealthIndicator {

  private static final Logger logger = LoggerFactory.getLogger(KafkaHealthCheck.class);

  private KafkaAdmin kafkaAdmin;

  private Map<String, Object> kafkaConfig;

  @Value("${application.topic}")
  private String topicName;

  @Value(value = "${kafka.bootstrapAddress}")
  private String bootstrapAddress;

  public KafkaHealthCheck(KafkaAdmin kafkaAdmin) {
    this.kafkaAdmin = kafkaAdmin;

  }

  @PostConstruct
  public void setUpAdminClient() {
    kafkaConfig = new HashMap<>();
    kafkaConfig.putAll(kafkaAdmin.getConfig());
    kafkaConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
  }

  @Override
  public Health health() {
    Long start = System.currentTimeMillis();
    try (AdminClient adminClient = AdminClient.create(kafkaConfig)) {

      DescribeClusterOptions describeClusterOptions = new DescribeClusterOptions().timeoutMs(2000);
      adminClient.describeCluster(describeClusterOptions);

      adminClient.describeConsumerGroups(List.of("topic")).all()
          .get(2, TimeUnit.SECONDS);

      Map<String, TopicDescription> topicDescriptionMap = adminClient
          .describeTopics(List.of(topicName)).all().get(2, TimeUnit.SECONDS);

      List<TopicPartitionInfo> partitions = topicDescriptionMap.get(topicName)
          .partitions();

      if (partitions == null || partitions.isEmpty()) {
        logger.warn(String
            .format("Kafka healthcheck failed - No partition found for topic: %s", topicName));
        return Health.down()
            .withDetail("Kafka healthcheck failed", "No partition found for topic: " + topicName)
            .build();
      } else {
        if (partitions.stream().anyMatch(p -> p.leader() == null)) {
          logger.warn(
              String.format("Kafka healthcheck failed - No partition leader found for topic: %s",
                  topicName));
          return Health.down().withDetail("Kafka healthcheck failed",
              "No partition leader found for topic: " + topicName).build();
        }
      }
    } catch (Exception e) {
      logger.warn("Kafka healthcheck failed", e);
      return Health.down()
          .withDetail("Kafka healthcheck failed", "Exception occurred during healthcheck").build();
    }
    System.out.println(System.currentTimeMillis() - start);
    return Health.up().build();
  }
}

以下是我在实施过程中遇到的问题：

1-KafkaAdmin在这个类中注入了我所有的配置（我使用的是SSL），除了“引导”。服务器'。我发现org.springframework.boot.autoconfigure.kafka.KafkaProperties将localhost:9092作为默认值，但应用程序配置在某种程度上没有覆盖它，而它对消费者和生产者都很好。我不知道为什么会这样，因此我必须在这里手动设置它

2-我在DescribeClusterOptions和describeConsumerGroups中添加了超时，但是这些超时似乎被完全忽略了。如果我手动关闭代理，healthcheck大约需要几分钟来报告错误

3-因为引导。servers错误，当我实际部署应用程序时，它几乎杀死了我的日志服务器，由org.apache.kafka.clients.NetworkClient生成的数百万条日志行表示Connection to node -1 could not be established. Broker may not be available.。我怎样才能阻止它再次发生？即使在经纪公司在运营期间倒闭的情况下

4-创建AdminClient时，即使是成功的运行状况检查也会生成大量日志行。它注销了它读取的所有配置和一系列其他语句。有没有可能把它最小化

5-总体而言，这是非常缓慢的。我试图计算只运行此健康检查所需的时间，平均约为1.5秒。有机会优化它吗