现在网上很难找到一篇有关HiveServer2 HA源码分析的文章，晚上有时间又耐不住从源码层面分析了下HiveServer2 HA负载均衡实现的原理。

HA有几种模式，一种是NameNode/ResourceManager/Flink等的FailOver Active/Standy切换的HA模型；一种则是HiveServer2 HA这种负载均衡的模型（应该还有其他吧？）。

HiveServer2 HA模型原理想起来应该很简单：

• Server端：在不同的机器上启动多个HiveServer2进程，每个进程都在Zookeeper指定namespace中加载该节点的ThriftServer的host/post等信息；
• Client端：从Zookeeper指定的namespace空间中随机选择出一个ThriftServer的host/port，连接该进程，就可以达到均衡的效果；
• 对于HA，则由Zookeeper的Watch机制完成，当发现某一台机器的HiveServer2进程挂掉的时候，会对Zookeeper相应的Znode做标记（删除？），这样下次Client端连接的时候就会把该节点自动过滤掉；

HiveServer2 Server端Zookeeper原理

HiveServer2的start()方法，在启动完毕HiveServer2各种Client之后，如果再配置中发现HIVE_SERVER2_SUPPORT_DYNAMIC_SERVICE_DISCOVERY dynamic service discovery模式，则会调用addServerInstanceToZooKeeper()将该ThriftServer的host/port信息加载至zookeeper中；

  @Override
  public synchronized void start() {
    super.start();
    // If we're supporting dynamic service discovery, we'll add the service uri for this
    // HiveServer2 instance to Zookeeper as a znode.
    HiveConf hiveConf = this.getHiveConf();
    if (hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_SUPPORT_DYNAMIC_SERVICE_DISCOVERY)) {
      try {
        addServerInstanceToZooKeeper(hiveConf);
      } catch (Exception e) {
        LOG.error("Error adding this HiveServer2 instance to ZooKeeper: ", e);
        throw new ServiceException(e);
      }
    }
    if (webServer != null) {
      try {
        webServer.start();
        LOG.info("Web UI has started on port " + webServer.getPort());
      } catch (Exception e) {
        LOG.error("Error starting Web UI: ", e);
        throw new ServiceException(e);
      }
    }
  }

addServerInstanceToZooKeeper()方法根据传入的HiveConf信息，将配置中的zookeeper HIVE_ZOOKEEPER_QUORUM加载至指定的HIVE_SERVER2_ZOOKEEPER_NAMESPACE空间中。

 private void addServerInstanceToZooKeeper(HiveConf hiveConf) throws Exception {
    String zooKeeperEnsemble = ZooKeeperHiveHelper.getQuorumServers(hiveConf);
    String rootNamespace = hiveConf.getVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE);
    String instanceURI = getServerInstanceURI();
    setUpZooKeeperAuth(hiveConf);
    int sessionTimeout =
        (int) hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT,
            TimeUnit.MILLISECONDS);
    int baseSleepTime =
        (int) hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_BASESLEEPTIME,
            TimeUnit.MILLISECONDS);
    int maxRetries = hiveConf.getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CONNECTION_MAX_RETRIES);
    // Create a CuratorFramework instance to be used as the ZooKeeper client
    // Use the zooKeeperAclProvider to create appropriate ACLs
    zooKeeperClient =
        CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
            .sessionTimeoutMs(sessionTimeout).aclProvider(zooKeeperAclProvider)
            .retryPolicy(new ExponentialBackoffRetry(baseSleepTime, maxRetries)).build();
    zooKeeperClient.start();
    // Create the parent znodes recursively; ignore if the parent already exists.
    try {
      zooKeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
          .forPath(ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + rootNamespace);
      LOG.info("Created the root name space: " + rootNamespace + " on ZooKeeper for HiveServer2");
    } catch (KeeperException e) {
      if (e.code() != KeeperException.Code.NODEEXISTS) {
        LOG.error("Unable to create HiveServer2 namespace: " + rootNamespace + " on ZooKeeper", e);
        throw e;
      }
    }
    // Create a znode under the rootNamespace parent for this instance of the server
    // Znode name: serverUri=host:port;version=versionInfo;sequence=sequenceNumber
    try {
      String pathPrefix =
          ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + rootNamespace
              + ZooKeeperHiveHelper.ZOOKEEPER_PATH_SEPARATOR + "serverUri=" + instanceURI + ";"
              + "version=" + HiveVersionInfo.getVersion() + ";" + "sequence=";
      String znodeData = "";
      if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_PUBLISH_CONFIGS)) {
        // HiveServer2 configs that this instance will publish to ZooKeeper,
        // so that the clients can read these and configure themselves properly.
        Map<String, String> confsToPublish = new HashMap<String, String>();
        addConfsToPublish(hiveConf, confsToPublish);
        // Publish configs for this instance as the data on the node
        znodeData = Joiner.on(';').withKeyValueSeparator("=").join(confsToPublish);
      } else {
        znodeData = instanceURI;
      }
      byte[] znodeDataUTF8 = znodeData.getBytes(Charset.forName("UTF-8"));
      znode =
          new PersistentEphemeralNode(zooKeeperClient,
              PersistentEphemeralNode.Mode.EPHEMERAL_SEQUENTIAL, pathPrefix, znodeDataUTF8);
      znode.start();
      // We'll wait for 120s for node creation
      long znodeCreationTimeout = 120;
      if (!znode.waitForInitialCreate(znodeCreationTimeout, TimeUnit.SECONDS)) {
        throw new Exception("Max znode creation wait time: " + znodeCreationTimeout + "s exhausted");
      }
      setDeregisteredWithZooKeeper(false);
      znodePath = znode.getActualPath();
      // Set a watch on the znode
      if (zooKeeperClient.checkExists().usingWatcher(new DeRegisterWatcher()).forPath(znodePath) == null) {
        // No node exists, throw exception
        throw new Exception("Unable to create znode for this HiveServer2 instance on ZooKeeper.");
      }
      LOG.info("Created a znode on ZooKeeper for HiveServer2 uri: " + instanceURI);
    } catch (Exception e) {
      LOG.error("Unable to create a znode for this server instance", e);
      if (znode != null) {
        znode.close();
      }
      throw (e);
    }
  }

Client连接原理

Hive Client端调用beeline连接HiveServer2 HA集群，如下：

beeline> !connect jdbc:hive2://hadoop710.lt.163.org:2181,hadoop711.lt.163.org:2181,hadoop712.lt.163.org:2181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2-lsm hadoop ""

Beeline调用org.apache.hive.jdbc.parseURL()解析jdbc连接字符串，判断serviceDiscoveryMode类型进行解析：

• 如果为zookeeper，则调用ZooKeeperHiveClientHelper解析；
• 否则解析出jdbc的host/port；

  private static void configureConnParams(JdbcConnectionParams connParams)
      throws JdbcUriParseException, ZooKeeperHiveClientException {
    String serviceDiscoveryMode =
        connParams.getSessionVars().get(JdbcConnectionParams.SERVICE_DISCOVERY_MODE);
    if ((serviceDiscoveryMode != null)
        && (JdbcConnectionParams.SERVICE_DISCOVERY_MODE_ZOOKEEPER
            .equalsIgnoreCase(serviceDiscoveryMode))) {
      // Set ZooKeeper ensemble in connParams for later use
      connParams.setZooKeeperEnsemble(joinStringArray(connParams.getAuthorityList(), ","));
      // Configure using ZooKeeper
      ZooKeeperHiveClientHelper.configureConnParams(connParams);
    } else {
      String authority = connParams.getAuthorityList()[0];
      URI jdbcURI = URI.create(URI_HIVE_PREFIX + "//" + authority);
      // Check to prevent unintentional use of embedded mode. A missing "/"
      // to separate the 'path' portion of URI can result in this.
      // The missing "/" common typo while using secure mode, eg of such url -
      // jdbc:hive2://localhost:10000;principal=hive/HiveServer2Host@YOUR-REALM.COM
      if (jdbcURI.getAuthority() != null) {
        String host = jdbcURI.getHost();
        int port = jdbcURI.getPort();
        if (host == null) {
          throw new JdbcUriParseException("Bad URL format. Hostname not found "
              + " in authority part of the url: " + jdbcURI.getAuthority()
              + ". Are you missing a '/' after the hostname ?");
        }
        // Set the port to default value; we do support jdbc url like:
        // jdbc:hive2://localhost/db
        if (port <= 0) {
          port = Integer.parseInt(Utils.DEFAULT_PORT);
        }
        connParams.setHost(jdbcURI.getHost());
        connParams.setPort(jdbcURI.getPort());
      }
    }
  }

org.apache.hive.jdbc.ZooKeeperHiveClientHelper中的configureConnParams()方法做如下事情：

• 从conf中获取所有的znodes节点；
• 去除已经尝试过的Znode节点(getCurrentHostZnodePath)；
• 然后在剩下的znode节点中随机的选择一个znode节点；
• 解析该znode节点的value信息，包含了jdbc所需要的host/port；

 static void configureConnParams(JdbcConnectionParams connParams)
      throws ZooKeeperHiveClientException {
    String zooKeeperEnsemble = connParams.getZooKeeperEnsemble();
    String zooKeeperNamespace =
        connParams.getSessionVars().get(JdbcConnectionParams.ZOOKEEPER_NAMESPACE);
    if ((zooKeeperNamespace == null) || (zooKeeperNamespace.isEmpty())) {
      zooKeeperNamespace = JdbcConnectionParams.ZOOKEEPER_DEFAULT_NAMESPACE;
    }
    List<String> serverHosts;
    Random randomizer = new Random();
    String serverNode;
    CuratorFramework zooKeeperClient =
        CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    try {
      zooKeeperClient.start();
      serverHosts = zooKeeperClient.getChildren().forPath("/" + zooKeeperNamespace);
      // Remove the znodes we've already tried from this list
      serverHosts.removeAll(connParams.getRejectedHostZnodePaths());
      if (serverHosts.isEmpty()) {
        throw new ZooKeeperHiveClientException(
            "Tried all existing HiveServer2 uris from ZooKeeper.");
      }
      // Now pick a server node randomly
      serverNode = serverHosts.get(randomizer.nextInt(serverHosts.size()));
      connParams.setCurrentHostZnodePath(serverNode);
      // Read data from the znode for this server node
      // This data could be either config string (new releases) or server end
      // point (old releases)
      String dataStr =
          new String(
              zooKeeperClient.getData().forPath("/" + zooKeeperNamespace + "/" + serverNode),
              Charset.forName("UTF-8"));
      Matcher matcher = kvPattern.matcher(dataStr);
      // If dataStr is not null and dataStr is not a KV pattern,
      // it must be the server uri added by an older version HS2
      if ((dataStr != null) && (!matcher.find())) {
        String[] split = dataStr.split(":");
        if (split.length != 2) {
          throw new ZooKeeperHiveClientException("Unable to read HiveServer2 uri from ZooKeeper: "
              + dataStr);
        }
        connParams.setHost(split[0]);
        connParams.setPort(Integer.parseInt(split[1]));
      } else {
        applyConfs(dataStr, connParams);
      }
    } catch (Exception e) {
      throw new ZooKeeperHiveClientException("Unable to read HiveServer2 configs from ZooKeeper", e);
    } finally {
      // Close the client connection with ZooKeeper
      if (zooKeeperClient != null) {
        zooKeeperClient.close();
      }
    }
  }

HA处理

当某一台host的HiveServerr2进程挂掉的时候，会调用HiveServer2.stop()方法，在该方法中会检测是否使用HIVE_SERVER2_SUPPORT_DYNAMIC_SERVICE_DISCOVERY模式，如果是该模式，则会将zookeeper相应相应实例删除掉；

    // Remove this server instance from ZooKeeper if dynamic service discovery is set
    if (hiveConf != null && hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_SUPPORT_DYNAMIC_SERVICE_DISCOVERY)) {
      try {
        removeServerInstanceFromZooKeeper();
      } catch (Exception e) {
        LOG.error("Error removing znode for this HiveServer2 instance from ZooKeeper.", e);
      }
    }

  /**
   * The watcher class which sets the de-register flag when the znode corresponding to this server
   * instance is deleted. Additionally, it shuts down the server if there are no more active client
   * sessions at the time of receiving a 'NodeDeleted' notification from ZooKeeper.
   */
  private class DeRegisterWatcher implements Watcher {
    @Override
    public void process(WatchedEvent event) {
      if (event.getType().equals(Watcher.Event.EventType.NodeDeleted)) {
        if (znode != null) {
          try {
            znode.close();
            LOG.warn("This HiveServer2 instance is now de-registered from ZooKeeper. "
                + "The server will be shut down after the last client sesssion completes.");
          } catch (IOException e) {
            LOG.error("Failed to close the persistent ephemeral znode", e);
          } finally {
            HiveServer2.this.setDeregisteredWithZooKeeper(true);
            // If there are no more active client sessions, stop the server
            if (cliService.getSessionManager().getOpenSessionCount() == 0) {
              LOG.warn("This instance of HiveServer2 has been removed from the list of server "
                  + "instances available for dynamic service discovery. "
                  + "The last client session has ended - will shutdown now.");
              HiveServer2.this.stop();
            }
          }
        }
      }
    }
  }