import org.I0Itec.zkclient.exception.ZkTimeoutException; //导入依赖的package包/类
/**
 * Connect to ZooKeeper.
 * 
 * @param maxMsToWaitUntilConnected
 * @param watcher
 * @throws ZkInterruptedException if the connection timed out due to thread interruption
 * @throws ZkTimeoutException if the connection timed out
 * @throws IllegalStateException if the connection timed out due to thread interruption
 */
public void connect(final long maxMsToWaitUntilConnected, Watcher watcher) throws ZkInterruptedException,
                                                                          ZkTimeoutException, IllegalStateException {
    boolean started = false;
    try {
        getEventLock().lockInterruptibly();
        setShutdownTrigger(false);
        _eventThread = new ZkEventThread(_connection.getServers());
        _eventThread.start();
        _connection.connect(watcher);

        LOG.debug("Awaiting connection to Zookeeper server");
        if (!waitUntilConnected(maxMsToWaitUntilConnected, TimeUnit.MILLISECONDS)) {
            throw new ZkTimeoutException("Unable to connect to zookeeper server within timeout: "
                                         + maxMsToWaitUntilConnected);
        }
        started = true;
    } catch (InterruptedException e) {
        States state = _connection.getZookeeperState();
        throw new IllegalStateException("Not connected with zookeeper server yet. Current state is " + state);
    } finally {
        getEventLock().unlock();

        // we should close the zookeeper instance, otherwise it would keep
        // on trying to connect
        if (!started) {
            close();
        }
    }
}
 

import org.I0Itec.zkclient.exception.ZkTimeoutException; //导入依赖的package包/类
@Test
public void start_failsOnInvalidZooKeeper() throws Exception {
  thrown.expect(ZkTimeoutException.class);
  thrown.expectMessage("Unable to connect to zookeeper server within timeout: 6000");

  Builder builder = builder("fail_invalid_zk").zookeeper("1.1.1.1");

  try (KafkaCollector collector = newKafkaTransport(builder, consumer)) {
  }
}
 

import org.I0Itec.zkclient.exception.ZkTimeoutException; //导入依赖的package包/类
@Override
protected void startUp() throws Exception {
  int tries = 0;
  do {
    KafkaConfig kafkaConfig = createKafkaConfig(properties);
    KafkaServer kafkaServer = createKafkaServer(kafkaConfig);
    try {
      kafkaServer.startup();
      server = kafkaServer;
    } catch (Exception e) {
      kafkaServer.shutdown();
      kafkaServer.awaitShutdown();

      Throwable rootCause = Throwables.getRootCause(e);
      if (rootCause instanceof ZkTimeoutException) {
        // Potentially caused by race condition bug described in TWILL-139.
        LOG.warn("Timeout when connecting to ZooKeeper from KafkaServer. Attempt number {}.", tries, rootCause);
      } else if (rootCause instanceof BindException) {
        LOG.warn("Kafka failed to bind to port {}. Attempt number {}.", kafkaConfig.port(), tries, rootCause);
      } else {
        throw e;
      }

      // Do a random sleep of < 200ms
      TimeUnit.MILLISECONDS.sleep(new Random().nextInt(200) + 1L);
    }
  } while (server == null && ++tries < startTimeoutRetries);

  if (server == null) {
    throw new IllegalStateException("Failed to start Kafka server after " + tries + " attempts.");
  }
}
 

import org.I0Itec.zkclient.exception.ZkTimeoutException; //导入依赖的package包/类
/**
 * Connect to ZooKeeper.
 *
 * @param maxMsToWaitUntilConnected
 * @param watcher
 * @throws ZkInterruptedException
 *             if the connection timed out due to thread interruption
 * @throws ZkTimeoutException
 *             if the connection timed out
 * @throws IllegalStateException
 *             if the connection timed out due to thread interruption
 */
public void connect(final long maxMsToWaitUntilConnected, Watcher watcher)
    throws ZkInterruptedException, ZkTimeoutException, IllegalStateException {
  boolean started = false;
  acquireEventLock();
  try {
    setShutdownTrigger(false);
    _eventThread = new ZkEventThread(_connection.getServers());
    _eventThread.start();
    _connection.connect(watcher);

    LOG.debug("Awaiting connection to Zookeeper server");
    if (!waitUntilConnected(maxMsToWaitUntilConnected, TimeUnit.MILLISECONDS)) {
      throw new ZkTimeoutException(
          "Unable to connect to zookeeper server within timeout: " + maxMsToWaitUntilConnected);
    }
    started = true;
  } finally {
    getEventLock().unlock();

    // we should close the zookeeper instance, otherwise it would keep
    // on trying to connect
    if (!started) {
      close();
    }
  }
}