Claim a resource.

@param zooKeeperConnection ZooKeeper connection to use.
@param poolSize            Size of the resource pool.
@param znode               Root znode of the ZooKeeper resource-pool.
@return A resource claim.
431644
public static ResourceClaim claimExpiring(ZooKeeperConnection zooKeeperConnection, int poolSize, String znode)
            throws IOException {
        return claimExpiring(zooKeeperConnection, poolSize, znode, DEFAULT_TIMEOUT);
    }
706602
private void waitForLastTxIdNode(AvatarZooKeeperClient zk, Configuration conf)
      throws Exception {
    // Gather session id and transaction id data.
    String address = conf.get(NameNode.DFS_NAMENODE_RPC_ADDRESS_KEY);
    long maxWaitTime = this.getMaxWaitTimeForWaitTxid();
    long start = System.currentTimeMillis();
    while (true) {
      if (System.currentTimeMillis() - start > maxWaitTime) {
        throw new IOException("No valid last txid znode found");
      }
      try {
        long sessionId = zk.getPrimarySsId(address, false);
        ZookeeperTxId zkTxId = zk.getPrimaryLastTxId(address, false);
        if (sessionId != zkTxId.getSessionId()) {
          LOG.warn("Session Id in the ssid node : " + sessionId
              + " does not match the session Id in the txid node : "
              + zkTxId.getSessionId() + " retrying...");
          Thread.sleep(retrySleep);
          continue;
        }
      } catch (Throwable e) {
        LOG.warn("Caught exception : " + e + " retrying ...");
        Thread.sleep(retrySleep);
        continue;
      }
      break;
    }
  }
1311057
private void setEntityManagerPoolSize(String poolSize) {

        setExtProperty(
            CmsDbPoolV11.KEY_DATABASE_POOL + '.' + getPool() + '.' + CmsDbPoolV11.KEY_ENTITY_MANAGER_POOL_SIZE,
            poolSize);
    }
902528
@Override
	protected ThriftConnection<T> getConnectionInternal() throws ThriftConnectionPoolException {
		ThriftConnection<T> result = pollConnection();
		// 如果立即获取连接失败 则换一个分区继续获取
		// TODO 设置当连接获取超时返回null？
		if (result == null) {
			if (this.pool.getThriftServerCount() == 0) {
				throw new ThriftConnectionPoolException("当前没有可用的服务器  无法获取连接");
			}
			int partition = (int) (Thread.currentThread().getId() % this.pool.thriftServerCount);

			ThriftConnectionPartition<T> thriftConnectionPartition = this.pool.partitions.get(partition);

			try {
				result = thriftConnectionPartition.poolFreeConnection(this.pool.connectionTimeoutInMs,
						TimeUnit.MILLISECONDS);
				if (result == null) {

					throw new ThriftConnectionPoolException("Timed out waiting for a free available connection.");
				}
			} catch (InterruptedException e) {
				throw new ThriftConnectionPoolException(e);
			}
		}
		return result;
	}
1243439
private ZooKeeper connect(String quorumAddresses) throws IOException {
        final CountDownLatch latch = new CountDownLatch(1);
        ZooKeeper zookeeper;

        // Connect to the quorum and wait for the successful connection callback.;
        zookeeper = new ZooKeeper(quorumAddresses, (int) SECONDS.toMillis(10), watchedEvent -> {
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                // Signal that the Zookeeper connection is established.
                latch.countDown();
            }
        });

        boolean successfullyConnected = false;
        try {
            successfullyConnected = latch.await(11, SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        if (!successfullyConnected) {
            throw new IOException(String.format(
                    "Connection to ZooKeeper quorum timed out after %d seconds.", CONNECTION_TIMEOUT));
        }

        return zookeeper;
    }
1244099
public boolean createNode(String path, byte[] value) throws ZooKeeperException {
        return _create(path, value, CreateMode.PERSISTENT);
    }
250331
void onConnect() throws InterruptedException, IOException
    {
        if (state.get() != NodeState.Fresh) {
            if (previousZKSessionStillActive()) {
                LOG.info("ZooKeeper session re-established before timeout.");
                return;
            }
            LOG.warn("Rejoined after session timeout. Forcing shutdown and clean startup.");
            ensureCleanStartup();
        }

        LOG.info("Connected to Zookeeper (ID: {}).", myNodeID);
        ZKUtils.ensureOrdasityPaths(zk, name, config.workUnitName, config.workUnitShortName);
        joinCluster();
        listener.onJoin(zk);
        if (watchesRegistered.compareAndSet(false, true)) {
            registerWatchers();
        }
        initialized.set(true);
        initializedLatch.countDown();
        setState(NodeState.Started);
        claimer.requestClaim();
        verifyIntegrity();

        balancingPolicy.onConnect();
        if (config.enableAutoRebalance) {
            scheduleRebalancing();
        }
    }
656308
private ThriftConnection<T> obtainRawInternalConnection(ThriftServerInfo serverInfo)
			throws ThriftConnectionPoolException {
		// 判断单服务还是多服务模式
		ThriftConnection<T> connection;
		if (this.thriftServiceType == ThriftServiceType.SINGLE_INTERFACE) {
			connection = new DefaultThriftConnection<>(serverInfo.getHost(), serverInfo.getPort(),
					this.connectionTimeOut, this.config.getThriftProtocol(), this.config.getClientClass());
		} else {
			connection = new MulitServiceThriftConnecion<>(serverInfo.getHost(), serverInfo.getPort(),
					this.connectionTimeOut, this.config.getThriftProtocol(), this.config.getThriftClientClasses());
		}
		return connection;
	}
999355
public ThriftConnection<T> obtainInternalConnection(ThriftConnectionHandle<T> thriftConnectionHandle)
			throws ThriftConnectionPoolException {
		boolean tryAgain;
		ThriftConnection<T> result;
		ThriftConnection<T> oldRawConnection = thriftConnectionHandle.getInternalConnection();
		ThriftServerInfo thriftServerInfo = thriftConnectionHandle.getThriftServerInfo();

		int acquireRetryAttempts = this.getConfig().getAcquireRetryAttempts();
		long acquireRetryDelayInMs = this.getConfig().getAcquireRetryDelayInMs();
		AcquireFailConfig acquireConfig = new AcquireFailConfig();
		acquireConfig.setAcquireRetryAttempts(new AtomicInteger(acquireRetryAttempts));
		acquireConfig.setAcquireRetryDelayInMs(acquireRetryDelayInMs);
		acquireConfig.setLogMessage("Failed to acquire connection to " + thriftServerInfo.toString());

		do {
			result = null;

			try {
				// 尝试获取原始连接
				result = this.obtainRawInternalConnection(thriftServerInfo);
				tryAgain = false;

				if (acquireRetryAttempts != this.getConfig().getAcquireRetryAttempts()) {
					logger.info("Successfully re-established connection to " + thriftServerInfo.toString());
				}

				thriftConnectionHandle.getConnectionPartition().getServerIsDown().set(false);

				thriftConnectionHandle.setInternalConnection(result);

			} catch (ThriftConnectionPoolException e) {
				logger.error(String.format("Failed to acquire connection to %s. Sleeping for %d ms. Attempts left: %d",
						thriftServerInfo.toString(), acquireRetryDelayInMs, acquireRetryAttempts), e);

				try {
					if (acquireRetryAttempts > 0) {
						Thread.sleep(acquireRetryDelayInMs);
					}
					tryAgain = (acquireRetryAttempts--) > 0;
				} catch (InterruptedException e1) {
					tryAgain = false;
				}

				if (!tryAgain) {
					if (oldRawConnection != null) {
						try {
							oldRawConnection.close();
						} catch (IOException e1) {
							throw new ThriftConnectionPoolException(e1);
						}
					}
					thriftConnectionHandle.setInternalConnection(oldRawConnection);
					throw e;
				}

			}
		} while (tryAgain);
		return result;
	}
901693
public static void createPersistentZKNodes(ZooKeeper zk) {
        LinkedList<ZKUtil.StringCallback> callbacks = new LinkedList<ZKUtil.StringCallback>();
        for (int i=0; i < VoltZK.ZK_HIERARCHY.length; i++) {
            ZKUtil.StringCallback cb = new ZKUtil.StringCallback();
            callbacks.add(cb);
            zk.create(VoltZK.ZK_HIERARCHY[i], null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, cb, null);
        }
        for (ZKUtil.StringCallback cb : callbacks) {
            try {
                cb.get();
            } catch (org.apache.zookeeper_voltpatches.KeeperException.NodeExistsException e) {
                // this is an expected race.
            } catch (Exception e) {
                VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
            }
        }
    }
844907
public synchronized long get() throws IOException {
    try {
      lastZNodeStat = zooKeeper.exists(fullyQualifiedZNode, false);
      if (lastZNodeStat == null) {
        return -1;
      }
      byte[] data =
          zooKeeper.getData(fullyQualifiedZNode, false, lastZNodeStat);
      WritableUtil.readWritableFromByteArray(data, maxTxIdWritable);
      return maxTxIdWritable.get();
    } catch (KeeperException e) {
      keeperException("Unrecoverable ZooKeeper error reading " +
          fullyQualifiedZNode, e);
      return -1; // This is never reached
    } catch (InterruptedException e) {
      interruptedException("Interrupted reading " +
          fullyQualifiedZNode, e);
      return -1; // This is never reached
    }
  }
132740
private ZooKeeper createZooKeeper() throws IOException {
    ZooKeeper zk = new ZooKeeper(zkStr, sessionTimeout, wrapWatcher(this));
    for (Map.Entry<String, byte[]> authInfo : authInfos.entries()) {
      zk.addAuthInfo(authInfo.getKey(), authInfo.getValue());
    }
    return zk;
  }
1121407
@Override
    public SimpleZooKeeperClient doConnect() throws IOException, InterruptedException {
        LOG.info("Connecting to zookeeper server - " + zooConf.getOnlineServerList());
        return new SimpleZooKeeperClient(zooConf.getOnlineServerList(),
                zooConf.getOnlineDigestAuth(), new ServerServiceWatcher(),
                zooConf.getSessionTimeout(), zooConf.getConnTimeout());
    }
118159
public int getConnectionPoolSize(URL url) {
        RequestBuilder req = new RequestBuilder();
        req.host = url.getHost();
        req.port = url.getPort() < 0 ? url.getDefaultPort() : url.getPort();
        return _getPoolSize(req);
    }
1079443
static String getPrimaryRegistration(Configuration startupConf,
      Configuration conf, String fsname) throws IOException {
    AvatarZooKeeperClient zk = null;
    int maxTries = startupConf.getInt("dfs.avatarnode.zk.retries", 3);
    for (int i = 0; i < maxTries; i++) {
      try {
        zk = new AvatarZooKeeperClient(conf, null, false);
        String zkRegistry = zk.getPrimaryAvatarAddress(fsname, new Stat(),
            false);
        return zkRegistry;
      } catch (Exception e) {
        LOG.error(
            "Got Exception when reading primary registration. Will retry...", e);
      } finally {
        shutdownZkClient(zk);
      }
    }
    throw new IOException("Cannot connect to zk");
  }
634945
protected Object retryOperation(ZooKeeperOperation operation) throws KeeperException, InterruptedException {
        KeeperException exception = null;
        for (int i = 0; i < retryCount; i++) {
            try {
                return operation.execute();
            } catch (KeeperException.SessionExpiredException e) {
                LOG.warn("Session expired for: " + zookeeper + " so reconnecting due to: " + e, e);
                throw e;
            } catch (KeeperException.ConnectionLossException e) {
                if (exception == null) {
                    exception = e;
                }
                LOG.debug("Attempt " + i + " failed with connection loss so " +
                              "attempting to reconnect: " + e, e);
                retryDelay(i);
            }
        }
        throw exception;
    }
323472
public void configMutliCluster(ZooKeeper zk) {
        if (_serversList.size() == 1) {
            return;
        }
        String cluster1 = _serversList.get(0);
        try {
            if (_serversList.size() > 1) {
                // 强制的声明accessible
                ReflectionUtils.makeAccessible(clientCnxnField);
                ReflectionUtils.makeAccessible(hostProviderField);
                ReflectionUtils.makeAccessible(serverAddressesField);

                // 添加第二组集群列表
                for (int i = 1; i < _serversList.size(); i++) {
                    String cluster = _serversList.get(i);
                    // 强制获取zk中的地址信息
                    ClientCnxn cnxn = (ClientCnxn) ReflectionUtils.getField(clientCnxnField, zk);
                    HostProvider hostProvider = (HostProvider) ReflectionUtils.getField(hostProviderField, cnxn);
                    List<InetSocketAddress> serverAddrs = (List<InetSocketAddress>) ReflectionUtils.getField(serverAddressesField,
                        hostProvider);
                    // 添加第二组集群列表
                    serverAddrs.addAll(new ConnectStringParser(cluster).getServerAddresses());
                }
            }
        } catch (Exception e) {
            try {
                if (zk != null) {
                    zk.close();
                }
            } catch (InterruptedException ie) {
                // ignore interrupt
            }
            throw new ZkException("zookeeper_create_error, serveraddrs=" + cluster1, e);
        }

    }