HIVE-8723: Set reasonable connection timeout for CuratorFramework Zoo…

…Keeper clients in Hive (Vaibhav Gumashta reviewed by Thejas Nair)

git-svn-id: https://svn.apache.org/repos/asf/hive/trunk@1636796 13f79535-47bb-0310-9956-ffa450edef68

itests/hive-unit/src/test/java/org/apache/hadoop/hive/thrift/TestZooKeeperTokenStore.java

@@ -43,9 +43,6 @@ public class TestZooKeeperTokenStore extends TestCase {
   private CuratorFramework zkClient = null;
   private int zkPort = -1;
   private ZooKeeperTokenStore ts;
-  // connect timeout large enough for slower test environments
-  private final int connectTimeoutMillis = 30000;
-  private final int sessionTimeoutMillis = 3000;
 
   @Override
   protected void setUp() throws Exception {
@@ -55,10 +52,9 @@ protected void setUp() throws Exception {
     }
     this.zkCluster = new MiniZooKeeperCluster();
     this.zkPort = this.zkCluster.startup(zkDataDir);
-
-    this.zkClient = CuratorFrameworkFactory.builder().connectString("localhost:" + zkPort)
-        .sessionTimeoutMs(sessionTimeoutMillis).connectionTimeoutMs(connectTimeoutMillis)
-        .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
+    this.zkClient =
+        CuratorFrameworkFactory.builder().connectString("localhost:" + zkPort)
+            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
     this.zkClient.start();
   }
 
@@ -74,15 +70,9 @@ protected void tearDown() throws Exception {
 
   private Configuration createConf(String zkPath) {
     Configuration conf = new Configuration();
-    conf.set(
-        HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR,
-        "localhost:" + this.zkPort);
-    conf.set(
-        HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ZNODE,
-        zkPath);
-    conf.setLong(
-        HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_TIMEOUTMILLIS,
-        connectTimeoutMillis);
+    conf.set(HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR, "localhost:"
+        + this.zkPort);
+    conf.set(HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ZNODE, zkPath);
     return conf;
   }
 

jdbc/src/java/org/apache/hive/jdbc/Utils.java

@@ -100,9 +100,6 @@ public static class JdbcConnectionParams {
     static final String ZOOKEEPER_DEFAULT_NAMESPACE = "hiveserver2";
 
     // Non-configurable params:
-    // ZOOKEEPER_SESSION_TIMEOUT is not exposed as client configurable
-    static final int ZOOKEEPER_SESSION_TIMEOUT = 600 * 1000;
-    static final int ZOOKEEPER_CONNECTION_TIMEOUT = -1;
     // Currently supports JKS keystore format
     static final String SSL_TRUST_STORE_TYPE = "JKS";
 

jdbc/src/java/org/apache/hive/jdbc/ZooKeeperHiveClientHelper.java

@@ -63,8 +63,6 @@ static String getNextServerUriFromZooKeeper(JdbcConnectionParams connParams)
     String serverNode;
     CuratorFramework zooKeeperClient =
         CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
-            .sessionTimeoutMs(JdbcConnectionParams.ZOOKEEPER_SESSION_TIMEOUT)
-            .connectionTimeoutMs(JdbcConnectionParams.ZOOKEEPER_CONNECTION_TIMEOUT)
             .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
     zooKeeperClient.start();
     try {

service/src/java/org/apache/hive/service/server/HiveServer2.java

@@ -143,9 +143,6 @@ public List<ACL> getAclForPath(String path) {
    * @throws Exception
    */
   private void addServerInstanceToZooKeeper(HiveConf hiveConf) throws Exception {
-    int zooKeeperSessionTimeout =
-        hiveConf.getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT);
-    int connectTimeoutMillis = -1;
     String zooKeeperEnsemble = ZooKeeperHiveHelper.getQuorumServers(hiveConf);
     String rootNamespace = hiveConf.getVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE);
     String instanceURI = getServerInstanceURI(hiveConf);
@@ -155,7 +152,6 @@ private void addServerInstanceToZooKeeper(HiveConf hiveConf) throws Exception {
     // Use the zooKeeperAclProvider to create appropriate ACLs
     zooKeeperClient =
         CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
-            .sessionTimeoutMs(zooKeeperSessionTimeout).connectionTimeoutMs(connectTimeoutMillis)
             .aclProvider(zooKeeperAclProvider).retryPolicy(new ExponentialBackoffRetry(1000, 3))
             .build();
     zooKeeperClient.start();
@@ -345,14 +341,10 @@ private static void startHiveServer2() throws Throwable {
    */
   static void deleteServerInstancesFromZooKeeper(String versionNumber) throws Exception {
     HiveConf hiveConf = new HiveConf();
-    int zooKeeperSessionTimeout =
-        hiveConf.getIntVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT);
-    int connectTimeoutMillis = -1;
     String zooKeeperEnsemble = ZooKeeperHiveHelper.getQuorumServers(hiveConf);
     String rootNamespace = hiveConf.getVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE);
     CuratorFramework zooKeeperClient =
         CuratorFrameworkFactory.builder().connectString(zooKeeperEnsemble)
-            .sessionTimeoutMs(zooKeeperSessionTimeout).connectionTimeoutMs(connectTimeoutMillis)
             .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
     zooKeeperClient.start();
     List<String> znodePaths =

shims/common-secure/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java

@@ -62,8 +62,7 @@ public class ZooKeeperTokenStore implements DelegationTokenStore {
   private String rootNode = "";
   private volatile CuratorFramework zkSession;
   private String zkConnectString;
-  private final int zkSessionTimeout = 3000;
-  private int connectTimeoutMillis = -1;
+  private int connectTimeoutMillis;
   private List<ACL> newNodeAcl = Arrays.asList(new ACL(Perms.ALL, Ids.AUTH_IDS));
 
   /**
@@ -101,10 +100,10 @@ private CuratorFramework getSession() {
     if (zkSession == null || zkSession.getState() == CuratorFrameworkState.STOPPED) {
       synchronized (this) {
         if (zkSession == null || zkSession.getState() == CuratorFrameworkState.STOPPED) {
-          zkSession = CuratorFrameworkFactory.builder().connectString(zkConnectString)
-              .sessionTimeoutMs(zkSessionTimeout).connectionTimeoutMs(connectTimeoutMillis)
-              .aclProvider(aclDefaultProvider)
-              .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
+          zkSession =
+              CuratorFrameworkFactory.builder().connectString(zkConnectString)
+                  .connectionTimeoutMs(connectTimeoutMillis).aclProvider(aclDefaultProvider)
+                  .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
           zkSession.start();
         }
       }
@@ -431,12 +430,14 @@ public void close() throws IOException {
   @Override
   public void init(Object objectStore, ServerMode smode) {
     this.serverMode = smode;
-    zkConnectString = conf.get(
-        HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR, null);
+    zkConnectString =
+        conf.get(HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR, null);
     if (zkConnectString == null || zkConnectString.trim().isEmpty()) {
       // try alternate config param
-      zkConnectString = conf.get(
-          HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR_ALTERNATE, null);
+      zkConnectString =
+          conf.get(
+              HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR_ALTERNATE,
+              null);
       if (zkConnectString == null || zkConnectString.trim().isEmpty()) {
         throw new IllegalArgumentException("Zookeeper connect string has to be specifed through "
             + "either " + HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR
@@ -445,14 +446,17 @@ public void init(Object objectStore, ServerMode smode) {
             + WHEN_ZK_DSTORE_MSG);
       }
     }
-    connectTimeoutMillis = conf.getInt(
-        HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_TIMEOUTMILLIS, -1);
+    connectTimeoutMillis =
+        conf.getInt(
+            HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_TIMEOUTMILLIS,
+            CuratorFrameworkFactory.builder().getConnectionTimeoutMs());
     String aclStr = conf.get(HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ACL, null);
     if (StringUtils.isNotBlank(aclStr)) {
       this.newNodeAcl = parseACLs(aclStr);
     }
-    rootNode = conf.get(HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ZNODE,
-        HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ZNODE_DEFAULT) + serverMode;
+    rootNode =
+        conf.get(HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ZNODE,
+            HadoopThriftAuthBridge20S.Server.DELEGATION_TOKEN_STORE_ZK_ZNODE_DEFAULT) + serverMode;
 
     try {
       // Install the JAAS Configuration for the runtime