@Test(timeout = 120000) public void isKafkaReady() throws Exception { Map<String, String> config = new HashMap<>(); config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapBroker (SecurityProtocol.PLAINTEXT)); assertThat(ClusterStatus.isKafkaReady(config, 3, 10000)) .isTrue(); }
public static boolean isKafkaReady( Map<String, String> config, int minBrokerCount, int timeoutMs ) { log.debug("Check if Kafka is ready: {}", config); AdminClient adminClient = AdminClient.create(new HashMap<String, Object>(config)); long begin = System.currentTimeMillis(); long remainingWaitMs = timeoutMs; Collection<Node> brokers = new ArrayList<>(); while (remainingWaitMs > 0) { 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); } 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; }
ClusterStatus { public static boolean isKafkaReady( Map<String, String> config, int minBrokerCount, int timeoutMs ) { log.debug("Check if Kafka is ready: {}", config); AdminClient adminClient = AdminClient.create(new HashMap<String, Object>(config)); long begin = System.currentTimeMillis(); long remainingWaitMs = timeoutMs; Collection<Node> brokers = new ArrayList<>(); while (remainingWaitMs > 0) { 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); } 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; } }
ClusterStatus { public static boolean isKafkaReady( Map<String, String> config, int minBrokerCount, int timeoutMs ) { log.debug("Check if Kafka is ready: {}", config); AdminClient adminClient = AdminClient.create(new HashMap<String, Object>(config)); long begin = System.currentTimeMillis(); long remainingWaitMs = timeoutMs; Collection<Node> brokers = new ArrayList<>(); while (remainingWaitMs > 0) { 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); } 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; }  }
ClusterStatus { public static boolean isKafkaReady( Map<String, String> config, int minBrokerCount, int timeoutMs ) { log.debug("Check if Kafka is ready: {}", config); AdminClient adminClient = AdminClient.create(new HashMap<String, Object>(config)); long begin = System.currentTimeMillis(); long remainingWaitMs = timeoutMs; Collection<Node> brokers = new ArrayList<>(); while (remainingWaitMs > 0) { 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); } 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; }  static boolean isZookeeperReady(String zkConnectString, int timeoutMs); static boolean isKafkaReady(
      Map<String, String> config,
      int minBrokerCount,
      int timeoutMs
  ); static Map<String, String> getKafkaEndpointFromZookeeper(
      String zkConnectString,
      int timeoutMs
  ); }
ClusterStatus { public static boolean isKafkaReady( Map<String, String> config, int minBrokerCount, int timeoutMs ) { log.debug("Check if Kafka is ready: {}", config); AdminClient adminClient = AdminClient.create(new HashMap<String, Object>(config)); long begin = System.currentTimeMillis(); long remainingWaitMs = timeoutMs; Collection<Node> brokers = new ArrayList<>(); while (remainingWaitMs > 0) { 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); } 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; }  static boolean isZookeeperReady(String zkConnectString, int timeoutMs); static boolean isKafkaReady(
      Map<String, String> config,
      int minBrokerCount,
      int timeoutMs
  ); static Map<String, String> getKafkaEndpointFromZookeeper(
      String zkConnectString,
      int timeoutMs
  ); static final String JAVA_SECURITY_AUTH_LOGIN_CONFIG; static final String BROKERS_IDS_PATH; static final int BROKER_METADATA_REQUEST_BACKOFF_MS; }