@Override
  public void tryBeLeader(Configuration conf, TableName table) throws LeaderException {
    requireNonNull(conf, "conf is null");
    requireNonNull(table, "table is null");
    this.enforceMutex = conf.getBoolean(HIVE_TXN_ENFORCE_AUX_MUTEX, true);
    if (store == null) {
      store = TxnUtils.getTxnStore(conf);
    }
    LockComponent component = new LockComponentBuilder()
        .setDbName(table.getDb())
        .setTableName(table.getTable())
        .setLock(LockType.EXCL_WRITE)
        .setOperationType(DataOperationType.NO_TXN)
        .build();
    List<LockComponent> components = new ArrayList<LockComponent>(1);
    components.add(component);

    boolean lockable = false;
    Exception recentException = null;
    long start = System.currentTimeMillis();
    LockRequest req = new LockRequest(components, userName, hostName);
    int numRetries = MetastoreConf.getIntVar(conf, MetastoreConf.ConfVars.LOCK_NUMRETRIES);
    long maxSleep = MetastoreConf.getTimeVar(conf,
        MetastoreConf.ConfVars.LOCK_SLEEP_BETWEEN_RETRIES, TimeUnit.MILLISECONDS);
    for (int i = 0; i < numRetries; i++) {
      try {
        LockResponse res = store.lock(req);
        if (res.getState() == LockState.WAITING || res.getState() == LockState.ACQUIRED) {
          lockable = true;
          doWork(res, conf, table);
          LOG.debug("Spent {}ms to lock the table {}, retries: {}", System.currentTimeMillis() - start, table, i);
          break;
        }
      } catch (NoSuchTxnException | TxnAbortedException e) {
        throw new AssertionError("This should not happen, we didn't open txn", e);
      } catch (MetaException e) {
        recentException = e;
        LOG.warn("Error while locking the table: {}, num retries: {}," +
                " max retries: {}, exception: {}", table, i, numRetries, e);
      }
      backoff(maxSleep);
    }
    if (!lockable) {
      throw new LeaderException("Error locking the table: " + table + " in " + numRetries +
          " retries, time spent: " + (System.currentTimeMillis() - start) + " ms",
          recentException);
    }
  }
Hive Lock based leader election.
If wins, the current instance becomes the leader,
and a heartbeat daemon will be spawned to refresh the lock before timeout.
If loses, a non-leader watcher will also be spawned to check the
lock periodically to see if he can grab the lock to be the leader.
The change of Leadership can be received by registering the
listeners through {@link LeaderElection#addStateListener}.
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaseLeaderElection.java