@Override
  public void setContext(HiveLockManagerCtx ctx) throws LockException {
    this.ctx = ctx;
    HiveConf conf = ctx.getConf();

    sleepTime = conf.getTimeVar(
        HiveConf.ConfVars.HIVE_LOCK_SLEEP_BETWEEN_RETRIES, TimeUnit.MILLISECONDS);
    numRetriesForLock = conf.getIntVar(HiveConf.ConfVars.HIVE_LOCK_NUMRETRIES);
    numRetriesForUnLock = conf.getIntVar(HiveConf.ConfVars.HIVE_UNLOCK_NUMRETRIES);

    try {
      curatorFramework = CuratorFrameworkSingleton.getInstance(conf);
      parent = conf.getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_NAMESPACE);
      try{
        curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/" +  parent, new byte[0]);
      } catch (Exception e) {
        // ignore if the parent already exists
        if (!(e instanceof KeeperException) || ((KeeperException)e).code() != KeeperException.Code.NODEEXISTS) {
          LOG.warn("Unexpected ZK exception when creating parent node /" + parent, e);
        }
      }
    } catch (Exception e) {
      LOG.error("Failed to create curatorFramework object: ", e);
      throw new LockException(ErrorMsg.ZOOKEEPER_CLIENT_COULD_NOT_BE_INITIALIZED.getMsg());
    }
  }
@param ctx  The lock manager context (containing the Hive configuration file)
Start the ZooKeeper client based on the zookeeper cluster specified in the conf.
ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java