@Test(timeout = 120000) public void zookeeperReadyWithBadConnectString() throws Exception { assertThat( ClusterStatus.isZookeeperReady("localhost:3245", 10000)) .isFalse(); }
public static boolean isZookeeperReady(String zkConnectString, int timeoutMs) { log.debug("Check if Zookeeper is ready: {} ", zkConnectString); ZooKeeper zookeeper = null; try { CountDownLatch waitForConnection = new CountDownLatch(1); boolean isSaslEnabled = false; if (System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null) != null) { isSaslEnabled = true; log.info( "SASL is enabled. java.security.auth.login.config={}", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG) ); } ZookeeperConnectionWatcher connectionWatcher = new ZookeeperConnectionWatcher(waitForConnection, isSaslEnabled); zookeeper = new ZooKeeper(zkConnectString, timeoutMs, connectionWatcher); boolean timedOut = !waitForConnection.await(timeoutMs, TimeUnit.MILLISECONDS); if (timedOut) { log.error( "Timed out waiting for connection to Zookeeper server [{}].", zkConnectString ); return false; } else if (!connectionWatcher.isSuccessful()) { log.error( "Error occurred while connecting to Zookeeper server[{}]. {} ", zkConnectString, connectionWatcher.getFailureMessage() ); return false; } else { return true; } } catch (Exception e) { log.error( "Error while waiting for Zookeeper client to connect to the server [{}].", zkConnectString, e ); return false; } finally { if (zookeeper != null) { try { zookeeper.close(); } catch (InterruptedException e) { log.error("Error while shutting down Zookeeper client.", e); Thread.currentThread().interrupt(); } } } }
ClusterStatus { public static boolean isZookeeperReady(String zkConnectString, int timeoutMs) { log.debug("Check if Zookeeper is ready: {} ", zkConnectString); ZooKeeper zookeeper = null; try { CountDownLatch waitForConnection = new CountDownLatch(1); boolean isSaslEnabled = false; if (System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null) != null) { isSaslEnabled = true; log.info( "SASL is enabled. java.security.auth.login.config={}", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG) ); } ZookeeperConnectionWatcher connectionWatcher = new ZookeeperConnectionWatcher(waitForConnection, isSaslEnabled); zookeeper = new ZooKeeper(zkConnectString, timeoutMs, connectionWatcher); boolean timedOut = !waitForConnection.await(timeoutMs, TimeUnit.MILLISECONDS); if (timedOut) { log.error( "Timed out waiting for connection to Zookeeper server [{}].", zkConnectString ); return false; } else if (!connectionWatcher.isSuccessful()) { log.error( "Error occurred while connecting to Zookeeper server[{}]. {} ", zkConnectString, connectionWatcher.getFailureMessage() ); return false; } else { return true; } } catch (Exception e) { log.error( "Error while waiting for Zookeeper client to connect to the server [{}].", zkConnectString, e ); return false; } finally { if (zookeeper != null) { try { zookeeper.close(); } catch (InterruptedException e) { log.error("Error while shutting down Zookeeper client.", e); Thread.currentThread().interrupt(); } } } } }
ClusterStatus { public static boolean isZookeeperReady(String zkConnectString, int timeoutMs) { log.debug("Check if Zookeeper is ready: {} ", zkConnectString); ZooKeeper zookeeper = null; try { CountDownLatch waitForConnection = new CountDownLatch(1); boolean isSaslEnabled = false; if (System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null) != null) { isSaslEnabled = true; log.info( "SASL is enabled. java.security.auth.login.config={}", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG) ); } ZookeeperConnectionWatcher connectionWatcher = new ZookeeperConnectionWatcher(waitForConnection, isSaslEnabled); zookeeper = new ZooKeeper(zkConnectString, timeoutMs, connectionWatcher); boolean timedOut = !waitForConnection.await(timeoutMs, TimeUnit.MILLISECONDS); if (timedOut) { log.error( "Timed out waiting for connection to Zookeeper server [{}].", zkConnectString ); return false; } else if (!connectionWatcher.isSuccessful()) { log.error( "Error occurred while connecting to Zookeeper server[{}]. {} ", zkConnectString, connectionWatcher.getFailureMessage() ); return false; } else { return true; } } catch (Exception e) { log.error( "Error while waiting for Zookeeper client to connect to the server [{}].", zkConnectString, e ); return false; } finally { if (zookeeper != null) { try { zookeeper.close(); } catch (InterruptedException e) { log.error("Error while shutting down Zookeeper client.", e); Thread.currentThread().interrupt(); } } } }  }
ClusterStatus { public static boolean isZookeeperReady(String zkConnectString, int timeoutMs) { log.debug("Check if Zookeeper is ready: {} ", zkConnectString); ZooKeeper zookeeper = null; try { CountDownLatch waitForConnection = new CountDownLatch(1); boolean isSaslEnabled = false; if (System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null) != null) { isSaslEnabled = true; log.info( "SASL is enabled. java.security.auth.login.config={}", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG) ); } ZookeeperConnectionWatcher connectionWatcher = new ZookeeperConnectionWatcher(waitForConnection, isSaslEnabled); zookeeper = new ZooKeeper(zkConnectString, timeoutMs, connectionWatcher); boolean timedOut = !waitForConnection.await(timeoutMs, TimeUnit.MILLISECONDS); if (timedOut) { log.error( "Timed out waiting for connection to Zookeeper server [{}].", zkConnectString ); return false; } else if (!connectionWatcher.isSuccessful()) { log.error( "Error occurred while connecting to Zookeeper server[{}]. {} ", zkConnectString, connectionWatcher.getFailureMessage() ); return false; } else { return true; } } catch (Exception e) { log.error( "Error while waiting for Zookeeper client to connect to the server [{}].", zkConnectString, e ); return false; } finally { if (zookeeper != null) { try { zookeeper.close(); } catch (InterruptedException e) { log.error("Error while shutting down Zookeeper client.", e); Thread.currentThread().interrupt(); } } } }  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 isZookeeperReady(String zkConnectString, int timeoutMs) { log.debug("Check if Zookeeper is ready: {} ", zkConnectString); ZooKeeper zookeeper = null; try { CountDownLatch waitForConnection = new CountDownLatch(1); boolean isSaslEnabled = false; if (System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null) != null) { isSaslEnabled = true; log.info( "SASL is enabled. java.security.auth.login.config={}", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG) ); } ZookeeperConnectionWatcher connectionWatcher = new ZookeeperConnectionWatcher(waitForConnection, isSaslEnabled); zookeeper = new ZooKeeper(zkConnectString, timeoutMs, connectionWatcher); boolean timedOut = !waitForConnection.await(timeoutMs, TimeUnit.MILLISECONDS); if (timedOut) { log.error( "Timed out waiting for connection to Zookeeper server [{}].", zkConnectString ); return false; } else if (!connectionWatcher.isSuccessful()) { log.error( "Error occurred while connecting to Zookeeper server[{}]. {} ", zkConnectString, connectionWatcher.getFailureMessage() ); return false; } else { return true; } } catch (Exception e) { log.error( "Error while waiting for Zookeeper client to connect to the server [{}].", zkConnectString, e ); return false; } finally { if (zookeeper != null) { try { zookeeper.close(); } catch (InterruptedException e) { log.error("Error while shutting down Zookeeper client.", e); Thread.currentThread().interrupt(); } } } }  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; }