@Override
    public void runInternal() {
      try {
        if (count ++ % 3 > 0) {
          // For WAITING AND NOT_ACQUIRED, re-check the lock at next cycle
          LockResponse res = store.checkLock(request);
          if (res.getState() == LockState.ACQUIRED) {
            // the current thread would be terminated by shutdownWatcher
            doWork(res, conf, tableName);
          } else if (res.getState() == LockState.ABORT) {
            reclaim();
          }
        } else {
          // In case the leader crashes, the lock it holds will become timeout eventually.
          // The AcidHouseKeeperService would not clean the corrupt lock until a new leader is elected,
          // however a leader candidate should hold that lock firstly in order to be the new leader,
          // a deadlock occurs in such case.
          // For all non-leader instances, they should try to clean timeout locks if possible to
          // avoid such problem.
          store.performTimeOuts();
        }
      } catch (NoSuchTxnException | TxnAbortedException e) {
        throw new AssertionError("This should not happen, we didn't open txn", e);
      } catch (NoSuchLockException e) {
        LOG.info("No such lock {} for NonLeaderWatcher, try to obtain the lock again...", lockId);
        reclaim();
      } catch (Exception e) {
        // Wait for next cycle.
        LOG.warn("CheckLock failed with exception: " + e.getMessage(), e);
      }
    }
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