private void queueGetRequestOnMasterThread(
      GetRequest req, HashSet<String> poolsToRedistribute, WmThreadSyncWork syncWork) {
    String poolName = userPoolMapping.mapSessionToPoolName(
        req.mappingInput, allowAnyPool, allowAnyPool ? pools.keySet() : null);
    if (poolName == null) {
      req.future.setException(new NoPoolMappingException(
          "Cannot find any pool mapping for " + req.mappingInput));
      returnSessionOnFailedReuse(req, syncWork, poolsToRedistribute);
      return;
    }
    PoolState pool = pools.get(poolName);
    if (pool == null) {
      req.future.setException(new AssertionError(poolName + " not found (internal error)."));
      returnSessionOnFailedReuse(req, syncWork, poolsToRedistribute);
      return;
    }

    PoolState oldPool = null;
    if (req.sessionToReuse != null) {
      // Given that we are trying to reuse, this session MUST be in some pool.sessions.
      // Kills that could have removed it must have cleared sessionToReuse.
      String oldPoolName = req.sessionToReuse.getPoolName();
      oldPool = pools.get(oldPoolName);
      RemoveSessionResult rr = checkAndRemoveSessionFromItsPool(
          req.sessionToReuse, poolsToRedistribute, true, false);
      if (rr != RemoveSessionResult.OK) {
        if (oldPool.metrics != null) {
          oldPool.metrics.removeRunningQueries(1);
        }
        // Abandon the reuse attempt.
        returnSessionOnFailedReuse(req, syncWork, null);
        req.sessionToReuse = null;
      } else if (pool.getTotalActiveSessions() + pool.queue.size() >= pool.queryParallelism) {
        if (oldPool.metrics != null) {
          oldPool.metrics.removeRunningQueries(1);
        }
        // One cannot simply reuse the session if there are other queries waiting; to maintain
        // fairness, we'll try to take a query slot instantly, and if that fails we'll return
        // this session back to the pool and give the user a new session later.
        returnSessionOnFailedReuse(req, syncWork, null);
        req.sessionToReuse = null;
      }
    }

    if (req.sessionToReuse != null) {
      // If we can immediately reuse a session, there's nothing to wait for - just return.
      req.sessionToReuse.setPoolName(poolName);
      req.sessionToReuse.setQueueName(yarnQueue);
      req.sessionToReuse.setQueryId(req.queryId);
      // Do not update metrics - we didn't update on removal.
      pool.sessions.add(req.sessionToReuse);
      if (pool != oldPool) {
        poolsToRedistribute.add(poolName);
      }
      req.future.set(req.sessionToReuse);
      return;
    }

    // Otherwise, queue the session and make sure we update this pool.
    pool.queue.addLast(req);
    if (pool.metrics != null) {
      pool.metrics.addQueuedQuery();
    }
    poolsToRedistribute.add(poolName);
  }
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