public static RecoverableZooKeeper connect(Configuration conf, String ensemble, Watcher watcher,
    final String identifier, ZKClientConfig zkClientConfig) throws IOException {
    if (ensemble == null) {
      throw new IOException("Unable to determine ZooKeeper ensemble");
    }
    int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
    if (LOG.isTraceEnabled()) {
      LOG.trace("{} opening connection to ZooKeeper ensemble={}", identifier, ensemble);
    }
    int retry = conf.getInt("zookeeper.recovery.retry", 3);
    int retryIntervalMillis = conf.getInt("zookeeper.recovery.retry.intervalmill", 1000);
    int maxSleepTime = conf.getInt("zookeeper.recovery.retry.maxsleeptime", 60000);
    int multiMaxSize = conf.getInt("zookeeper.multi.max.size", 1024 * 1024);
    return new RecoverableZooKeeper(ensemble, timeout, watcher, retry, retryIntervalMillis,
      maxSleepTime, identifier, multiMaxSize, zkClientConfig);
  }
Creates a new connection to ZooKeeper, pulling settings and ensemble config from the specified
configuration object using methods from {@link ZKConfig}. Sets the connection status monitoring
watcher to the specified watcher.
@param conf           configuration to pull ensemble and other settings from
@param watcher        watcher to monitor connection changes
@param ensemble       ZooKeeper servers quorum string
@param identifier     value used to identify this client instance.
@param zkClientConfig client specific configurations for this instance
@return connection to zookeeper
@throws IOException if unable to connect to zk or config problem
hbase-zookeeper/src/main/java/org/apache/hadoop/hbase/zookeeper/RecoverableZooKeeper.java