@Override
  public void init(Object hmsHandler, HadoopThriftAuthBridge.Server.ServerMode sMode) {
    this.serverMode = sMode;
    zkConnectString =
        conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR, null);
    if (zkConnectString == null || zkConnectString.trim().isEmpty()) {
      // try alternate config param
      zkConnectString = conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR_ALTERNATE, null);
      if (zkConnectString == null || zkConnectString.trim().isEmpty()) {
        throw new IllegalArgumentException("Zookeeper connect string has to be specified through either "
            + MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR + " or "
            + MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR_ALTERNATE + WHEN_ZK_DSTORE_MSG);
      }
      // If we use the alternate zk config (the global one)
      // we should also use the related properties from the global config
      zkConnectPort = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_CLIENT_PORT);
      connectTimeoutMillis = (int) MetastoreConf
          .getTimeVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);
      sslEnabled = MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_ENABLE);
      if (sslEnabled) {
        try {
          keyStoreLocation = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_KEYSTORE_LOCATION);
          keyStorePassword =
              MetastoreConf.getPassword(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_KEYSTORE_PASSWORD);
          keyStoreType =
              MetastoreConf.getVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_KEYSTORE_TYPE);
          trustStoreLocation =
              MetastoreConf.getVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_TRUSTSTORE_LOCATION);
          trustStorePassword =
              MetastoreConf.getPassword(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD);
          trustStoreType =
              MetastoreConf.getVar(conf, MetastoreConf.ConfVars.THRIFT_ZOOKEEPER_SSL_TRUSTSTORE_TYPE);
        } catch (IOException ex) {
          throw new RuntimeException("Failed to read zookeeper configuration passwords", ex);
        }
      }
    } else {
      connectTimeoutMillis =
          conf.getInt(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_CONNECT_TIMEOUTMILLIS,
              CuratorFrameworkFactory.builder().getConnectionTimeoutMs());
      sslEnabled = conf.getBoolean(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_SSL_ENABLE, false);
      if (sslEnabled) {
        try {
          keyStoreLocation = conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_KEYSTORE_LOCATION, "");
          char[] pwd = conf.getPassword(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_KEYSTORE_PASSWORD);
          keyStorePassword = pwd == null ? null : new String(pwd);
          keyStoreType = conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_KEYSTORE_TYPE, "");
          trustStoreLocation =
              conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_LOCATION, "");
          pwd = conf.getPassword(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_PASSWORD);
          trustStorePassword = pwd == null ? null : new String(pwd);
          trustStoreType = conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_TRUSTSTORE_TYPE, "");
        } catch (IOException ex) {
          throw new RuntimeException("Failed to read zookeeper configuration passwords", ex);
        }
      }
    }

    String aclStr = conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_ACL, null);
    this.newNodeAcl = StringUtils.isNotBlank(aclStr)? parseACLs(aclStr) : getDefaultAcl(conf);

    rootNode =
        conf.get(MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_ZNODE,
            MetastoreDelegationTokenManager.DELEGATION_TOKEN_STORE_ZK_ZNODE_DEFAULT) + serverMode;

    try {
      // Install the JAAS Configuration for the runtime
      setupJAASConfig(conf);
    } catch (IOException e) {
      throw new TokenStoreException("Error setting up JAAS configuration for zookeeper client "
          + e.getMessage(), e);
    }
    initClientAndPaths();
  }
Parse comma separated list of ACL entries to secure generated nodes, e.g.
<code>sasl:hive/host1@MY.DOMAIN:cdrwa,sasl:hive/host2@MY.DOMAIN:cdrwa</code>
@param aclString
@return ACL list
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/security/ZooKeeperTokenStore.java