private void processPoolChangesOnMasterThread(
      String poolName, boolean hasRequeues, WmThreadSyncWork syncWork) throws Exception {
    PoolState pool = pools.get(poolName);
    if (pool == null) return; // Might be from before the new resource plan.

    // 1. First, start the queries from the queue.
    int queriesToStart = Math.min(pool.queue.size(),
        pool.queryParallelism - pool.getTotalActiveSessions());

    if (queriesToStart > 0) {
      LOG.info("Starting {} queries in pool {}", queriesToStart, pool);
    }
    if (hasRequeues) {
      // Sort the queue - we may have put items here out of order.
      Collections.sort(pool.queue, GetRequest.ORDER_COMPARATOR);
    }
    for (int i = 0; i < queriesToStart; ++i) {
      GetRequest queueReq = pool.queue.pollFirst();
      if (pool.metrics != null) {
        pool.metrics.moveQueuedToRunning();
      }
      assert queueReq.sessionToReuse == null;
      // Note that in theory, we are guaranteed to have a session waiting for us here, but
      // the expiration, failures, etc. may cause one to be missing pending restart.
      // See SessionInitContext javadoc.
      SessionInitContext sw = new SessionInitContext(
          queueReq.future, poolName, queueReq.queryId, queueReq.wmContext, null);
      sw.start();
      // It is possible that all the async methods returned on the same thread because the
      // session with registry data and stuff was available in the pool.
      // If this happens, we'll take the session out here and "cancel" the init so we skip
      // processing the message that the successful init has queued for us.
      boolean isDone = sw.extractSessionAndCancelIfDone(pool.sessions, syncWork.pathsToDelete);
      if (!isDone) {
        pool.initializingSessions.add(sw);
      }
      // The user has already been notified of completion by SessionInitContext.
    }

    // 2. Then, update pool allocations.
    double totalAlloc = pool.updateAllocationPercentages();
    // We are calling this here because we expect the method to be completely async. We also don't
    // want this call itself to go on a thread because we want the percent-to-physics conversion
    // logic to be consistent between all the separate calls in one master thread processing round.
    // Note: If allocation manager does not have cluster state, it won't update anything. When the
    //       cluster state changes, it will notify us, and we'd update the queries again.
    int cpusAllocated = allocationManager.updateSessionsAsync(totalAlloc, pool.sessions);
    if (pool.metrics != null) {
      pool.metrics.setExecutors(cpusAllocated);
      if (cpusAllocated > 0) {
        // Update max executors now that cluster info is definitely available.
        pool.metrics.setMaxExecutors(allocationManager.translateAllocationToCpus(totalAlloc));
      }
    }
  }
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