test_class.declarator: kafka
test_class.modifier: private static
test_class.original_string: private static EmbeddedKafkaCluster kafka;
test_class.type: EmbeddedKafkaCluster
test_class.var_name: kafka
test_class.declarator: numBrokers = 3
test_class.modifier: private static
test_class.original_string: private static int numBrokers = 3;
test_class.type: int
test_class.var_name: numBrokers
test_class.declarator: numZookeeperPeers = 3
test_class.modifier: private static
test_class.original_string: private static int numZookeeperPeers = 3;
test_class.type: int
test_class.var_name: numZookeeperPeers
test_class.file: utility-belt/src/test/java/io/confluent/admin/utils/ClusterStatusTest.java
test_class.identifier: ClusterStatusTest
test_case.body: @Test(timeout = 120000)
    public void isKafkaReadyWaitFailureWithNoBroker() throws Exception {
      try {
        Map<String, String> config = new HashMap<>();
        config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:6789");
        assertThat(ClusterStatus.isKafkaReady(config, 3, 10000)).isFalse();
      } catch (Exception e) {
        fail("Unexpected error." + e.getMessage());
      }
    }
test_case.class_method_signature: ClusterStatusTest.isKafkaReadyWaitFailureWithNoBroker()
test_case.full_signature: @Test(timeout = 120000) public void isKafkaReadyWaitFailureWithNoBroker()
test_case.identifier: isKafkaReadyWaitFailureWithNoBroker
test_case.invocations: put isFalse assertThat isKafkaReady fail getMessage
test_case.modifiers: @Test(timeout = 120000) public
test_case.parameters: ()
test_case.return: void
test_case.signature: void isKafkaReadyWaitFailureWithNoBroker()
focal_class.declarator: log = LoggerFactory.getLogger(ClusterStatus.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(ClusterStatus.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: JAVA_SECURITY_AUTH_LOGIN_CONFIG = "java.security.auth.login.config"
focal_class.modifier: public static final
focal_class.original_string: public static final String JAVA_SECURITY_AUTH_LOGIN_CONFIG = "java.security.auth.login.config";
focal_class.type: String
focal_class.var_name: JAVA_SECURITY_AUTH_LOGIN_CONFIG
focal_class.declarator: BROKERS_IDS_PATH = "/brokers/ids"
focal_class.modifier: public static final
focal_class.original_string: public static final String BROKERS_IDS_PATH = "/brokers/ids";
focal_class.type: String
focal_class.var_name: BROKERS_IDS_PATH
focal_class.declarator: BROKER_METADATA_REQUEST_BACKOFF_MS = 1000
focal_class.modifier: public static final
focal_class.original_string: public static final int BROKER_METADATA_REQUEST_BACKOFF_MS = 1000;
focal_class.type: int
focal_class.var_name: BROKER_METADATA_REQUEST_BACKOFF_MS
focal_class.file: utility-belt/src/main/java/io/confluent/admin/utils/ClusterStatus.java
focal_class.identifier: ClusterStatus
focal_class.class_method_signature: ClusterStatus.isZookeeperReady(String zkConnectString, int timeoutMs)
focal_class.full_signature: public static boolean isZookeeperReady(String zkConnectString, int timeoutMs)
focal_class.identifier: isZookeeperReady
focal_class.modifiers: public static
focal_class.parameters: (String zkConnectString, int timeoutMs)
focal_class.return: boolean
focal_class.signature: boolean isZookeeperReady(String zkConnectString, int timeoutMs)
focal_class.class_method_signature: ClusterStatus.isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_class.full_signature: public static boolean isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_class.identifier: isKafkaReady
focal_class.modifiers: public static
focal_class.parameters: (
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_class.return: boolean
focal_class.signature: boolean isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_class.class_method_signature: ClusterStatus.getBrokerMetadataFromZookeeper(
        String zkConnectString,
        int timeoutMs
    )
focal_class.full_signature: private static List<String> getBrokerMetadataFromZookeeper(
        String zkConnectString,
        int timeoutMs
    )
focal_class.identifier: getBrokerMetadataFromZookeeper
focal_class.modifiers: private static
focal_class.parameters: (
        String zkConnectString,
        int timeoutMs
    )
focal_class.return: List<String>
focal_class.signature: List<String> getBrokerMetadataFromZookeeper(
        String zkConnectString,
        int timeoutMs
    )
focal_class.class_method_signature: ClusterStatus.getRawKafkaMetadataFromZK(ZooKeeper zookeeper, int timeoutMs)
focal_class.full_signature: private static List<String> getRawKafkaMetadataFromZK(ZooKeeper zookeeper, int timeoutMs)
focal_class.identifier: getRawKafkaMetadataFromZK
focal_class.modifiers: private static
focal_class.parameters: (ZooKeeper zookeeper, int timeoutMs)
focal_class.return: List<String>
focal_class.signature: List<String> getRawKafkaMetadataFromZK(ZooKeeper zookeeper, int timeoutMs)
focal_class.class_method_signature: ClusterStatus.isKafkaRegisteredInZookeeper(ZooKeeper zookeeper, int timeoutMs)
focal_class.full_signature: private static boolean isKafkaRegisteredInZookeeper(ZooKeeper zookeeper, int timeoutMs)
focal_class.identifier: isKafkaRegisteredInZookeeper
focal_class.modifiers: private static
focal_class.parameters: (ZooKeeper zookeeper, int timeoutMs)
focal_class.return: boolean
focal_class.signature: boolean isKafkaRegisteredInZookeeper(ZooKeeper zookeeper, int timeoutMs)
focal_class.class_method_signature: ClusterStatus.createZookeeperClient(String zkConnectString, int timeoutMs)
focal_class.full_signature: private static ZooKeeper createZookeeperClient(String zkConnectString, int timeoutMs)
focal_class.identifier: createZookeeperClient
focal_class.modifiers: private static
focal_class.parameters: (String zkConnectString, int timeoutMs)
focal_class.return: ZooKeeper
focal_class.signature: ZooKeeper createZookeeperClient(String zkConnectString, int timeoutMs)
focal_class.class_method_signature: ClusterStatus.sleep(long ms)
focal_class.full_signature: private static void sleep(long ms)
focal_class.identifier: sleep
focal_class.modifiers: private static
focal_class.parameters: (long ms)
focal_class.return: void
focal_class.signature: void sleep(long ms)
focal_class.class_method_signature: ClusterStatus.getKafkaEndpointFromZookeeper(
        String zkConnectString,
        int timeoutMs
    )
focal_class.full_signature: public static Map<String, String> getKafkaEndpointFromZookeeper(
        String zkConnectString,
        int timeoutMs
    )
focal_class.identifier: getKafkaEndpointFromZookeeper
focal_class.modifiers: public static
focal_class.parameters: (
        String zkConnectString,
        int timeoutMs
    )
focal_class.return: Map<String, String>
focal_class.signature: Map<String, String> getKafkaEndpointFromZookeeper(
        String zkConnectString,
        int timeoutMs
    )
focal_method.body: public static boolean isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    ) {
  
      log.debug("Check if Kafka is ready: {}", config);
  
      // Need to copy because `config` is Map<String, String> and `create` expects Map<String, Object>
      AdminClient adminClient = AdminClient.create(new HashMap<String, Object>(config));
  
      long begin = System.currentTimeMillis();
      long remainingWaitMs = timeoutMs;
      Collection<Node> brokers = new ArrayList<>();
      while (remainingWaitMs > 0) {
  
        // describeCluster does not wait for all brokers to be ready before returning the brokers.
        // So, wait until expected brokers are present or the time out expires.
        try {
          brokers = adminClient.describeCluster(new DescribeClusterOptions().timeoutMs(
                  (int) Math.min(Integer.MAX_VALUE, remainingWaitMs))).nodes().get();
          log.debug("Broker list: {}", (brokers != null ? brokers : "[]"));
          if ((brokers != null) && (brokers.size() >= minBrokerCount)) {
            return true;
          }
        } catch (Exception e) {
          log.error("Error while getting broker list.", e);
          // Swallow exceptions because we want to retry until timeoutMs expires.
        }
  
        sleep(Math.min(BROKER_METADATA_REQUEST_BACKOFF_MS, remainingWaitMs));
  
        log.info(
            "Expected {} brokers but found only {}. Trying to query Kafka for metadata again ...",
            minBrokerCount,
            brokers == null ? 0 : brokers.size()
        );
        long elapsed = System.currentTimeMillis() - begin;
        remainingWaitMs = timeoutMs - elapsed;
      }
  
      log.error(
          "Expected {} brokers but found only {}. Brokers found {}.",
          minBrokerCount,
          brokers == null ? 0 : brokers.size(),
          brokers != null ? brokers : "[]"
      );
  
      return false;
    }
focal_method.class_method_signature: ClusterStatus.isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_method.full_signature: public static boolean isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_method.identifier: isKafkaReady
focal_method.invocations: debug create currentTimeMillis get nodes describeCluster timeoutMs min debug size error sleep min info size currentTimeMillis error size
focal_method.modifiers: public static
focal_method.parameters: (
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )
focal_method.return: boolean
focal_method.signature: boolean isKafkaReady(
        Map<String, String> config,
        int minBrokerCount,
        int timeoutMs
    )