private void processDelayedMovesForPool(final String poolName, final HashSet<String> poolsToRedistribute, final Map<WmTezSession, WmEvent> recordMoveEvents,
      WmThreadSyncWork syncWork, IdentityHashMap<WmTezSession, GetRequest> toReuse) {
    long currentTime = System.currentTimeMillis();
    PoolState pool = pools.get(poolName);
    int movedCount = 0;
    int queueSize = pool.queue.size();
    int remainingCapacity = pool.queryParallelism - pool.getTotalActiveSessions();
    int delayedMovesToProcess = (queueSize > remainingCapacity) ? (queueSize - remainingCapacity) : 0;
    Iterator<MoveSession> iter = pool.delayedMoveSessions.iterator();
    while (iter.hasNext()) {
      MoveSession moveSession = iter.next();
      MoveSessionResult result;

      //Discard the delayed move if invalid
      if (!validDelayedMove(moveSession, pool, poolName)) {
        iter.remove();
        continue;
      }

      // Process the delayed move if
      // 1. The delayed move has timed out or
      // 2. The destination pool has freed up or
      // 3. If the source pool has incoming requests and we need to free up capacity in the source pool
      // to accommodate these requests.
      if ((movedCount < delayedMovesToProcess) || (capacityAvailable(moveSession.destPool))
          || ((delayedMoveTimeOutSec > 0) && ((currentTime - moveSession.startTime) >= delayedMoveTimeOutSec * 1000))){
        LOG.info("Processing delayed move {} for pool {}", moveSession, poolName);
        result = handleMoveSessionOnMasterThread(moveSession, syncWork, poolsToRedistribute, toReuse, recordMoveEvents,
            false);
        LOG.info("Result of processing delayed move {} for pool {}: {}", moveSession, poolName, result);
        iter.remove();
        if ((result == MoveSessionResult.OK) || (result == MoveSessionResult.KILLED)) {
          movedCount++;
        }
      }
    }
  }
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