private MoveSessionResult handleMoveSessionOnMasterThread(final MoveSession moveSession,
      final WmThreadSyncWork syncWork,
      final HashSet<String> poolsToRedistribute,
      final Map<WmTezSession, GetRequest> toReuse,
      final Map<WmTezSession, WmEvent> recordMoveEvents,
      final boolean convertToDelayedMove) {
    String destPoolName = moveSession.destPool;
    LOG.info("Handling move session event: {}, Convert to Delayed Move: {}", moveSession, convertToDelayedMove);
    if (validMove(moveSession.srcSession, destPoolName)) {
      String srcPoolName = moveSession.srcSession.getPoolName();
      PoolState srcPool = pools.get(srcPoolName);
      boolean capacityAvailableInDest = capacityAvailable(destPoolName);
      // If delayed move is set to true and if destination pool doesn't have enough capacity, don't kill the query.
      // Let the query run in source pool. Add the session to the source pool's delayed move sessions.
      if (convertToDelayedMove && !capacityAvailableInDest) {
        srcPool.delayedMoveSessions.add(moveSession);
        moveSession.srcSession.setDelayedMove(true);
        LOG.info("Converting Move: {} to a delayed move. Since destination pool {} is full, running in source pool {}"
            + " as long as possible.", moveSession, destPoolName, srcPoolName);
        moveSession.future.set(false);
        return MoveSessionResult.CONVERTED_TO_DELAYED_MOVE;
      }
      WmEvent moveEvent = new WmEvent(WmEvent.EventType.MOVE);
      // remove from src pool
      if (moveSession.srcSession.isDelayedMove()) {
        moveSession.srcSession.setDelayedMove(false);
      }
      RemoveSessionResult rr = checkAndRemoveSessionFromItsPool(
          moveSession.srcSession, poolsToRedistribute, true, true);
      if (rr == RemoveSessionResult.OK) {
        // check if there is capacity in dest pool, if so move else kill the session
        if (capacityAvailableInDest) {
          // add to destination pool
          Boolean added = checkAndAddSessionToAnotherPool(
              moveSession.srcSession, destPoolName, poolsToRedistribute);
          if (added != null && added) {
            moveSession.future.set(true);
            recordMoveEvents.put(moveSession.srcSession, moveEvent);
            return MoveSessionResult.OK;
          } else {
            LOG.error("Failed to move session: {}. Session is not added to destination.", moveSession);
          }
        } else {
          WmTezSession session = moveSession.srcSession;
          KillQueryContext killQueryContext = new KillQueryContext(session, "Destination pool " + destPoolName +
             " is full. Killing query.");
          resetAndQueueKill(syncWork.toKillQuery, killQueryContext, toReuse);
          moveSession.future.set(false);
          return MoveSessionResult.KILLED;
        }
      } else {
        LOG.error("Failed to move session: {}. Session is not removed from its pool.", moveSession);
      }
    } else {
      LOG.error("Validation failed for move session: {}. Invalid move or session/pool got removed.", moveSession);
    }
    moveSession.future.set(false);
    return MoveSessionResult.ERROR;
  }
This is the main method of the master thread the processes one set of events.
Be mindful of the fact that events can be queued while we are processing events, so
in addition to making sure we keep the current set consistent (e.g. no need to handle
update errors for a session that should already be destroyed), this needs to guard itself
against the future iterations - e.g. what happens if we kill a query due to plan change,
but the DAG finished before the kill happens and the user queues a "return" event? Etc.
DO NOT block for a long time in this method.
@param e Input events.
@param syncWork Output tasks that cannot be called via async methods.
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java