private void applyNewResourcePlanOnMasterThread(
    EventState e, WmThreadSyncWork syncWork, HashSet<String> poolsToRedistribute) {
    int totalQueryParallelism = 0;
    WMFullResourcePlan plan = e.resourcePlanToApply;
    if (plan == null) {
      // NULL plan means WM is disabled via a command; it could still be reenabled.
      LOG.info("Disabling workload management because the resource plan has been removed");
      this.rpName = null;
      this.defaultPool = null;
      this.userPoolMapping = new UserPoolMapping(null, null);
    } else {
      this.rpName = plan.getPlan().getName();
      this.defaultPool = plan.getPlan().getDefaultPoolPath();
      this.userPoolMapping = new UserPoolMapping(plan.getMappings(), defaultPool);
    }
    // Note: we assume here that plan has been validated beforehand, so we don't verify
    //       that fractions or query parallelism add up, etc.
    Map<String, PoolState> oldPools = pools;
    pools = new HashMap<>();

    ArrayList<List<WMPool>> poolsByLevel = new ArrayList<>();
    if (plan != null) {
      // For simplicity, to always have parents while storing pools in a flat structure, we'll
      // first distribute them by levels, then add level by level.
      for (WMPool pool : plan.getPools()) {
        String fullName = pool.getPoolPath();
        int ix = StringUtils.countMatches(fullName, POOL_SEPARATOR_STR);
        while (poolsByLevel.size() <= ix) {
          poolsByLevel.add(new LinkedList<WMPool>()); // We expect all the levels to have items.
        }
        poolsByLevel.get(ix).add(pool);
      }
    }
    for (int level = 0; level < poolsByLevel.size(); ++level) {
      List<WMPool> poolsOnLevel = poolsByLevel.get(level);
      for (WMPool pool : poolsOnLevel) {
        String fullName = pool.getPoolPath();
        int qp = pool.getQueryParallelism();
        double fraction = pool.getAllocFraction();
        if (level > 0) {
          String parentName = fullName.substring(0, fullName.lastIndexOf(POOL_SEPARATOR));
          PoolState parent = pools.get(parentName);
          fraction = parent.finalFraction * fraction;
          parent.finalFractionRemaining -= fraction;
        }
        PoolState state = oldPools == null ? null : oldPools.remove(fullName);
        if (state == null) {
          state = new PoolState(fullName, qp, fraction, pool.getSchedulingPolicy(), metricsSystem);
        } else {
          // This will also take care of the queries if query parallelism changed.
          state.update(qp, fraction, syncWork, e, pool.getSchedulingPolicy());
          poolsToRedistribute.add(fullName);
        }
        state.setTriggers(new LinkedList<Trigger>());
        LOG.info("Adding Hive pool: " + state);
        pools.put(fullName, state);
        totalQueryParallelism += qp;
      }
    }
    for (PoolState pool : pools.values()) {
      if (pool.metrics != null) {
        pool.metrics.setMaxExecutors(
            allocationManager.translateAllocationToCpus(pool.finalFractionRemaining));
      }
    }
    // TODO: in the current impl, triggers are added to RP. For tez, no pool triggers (mapping between trigger name and
    // pool name) will exist which means all triggers applies to tez. For LLAP, pool triggers has to exist for attaching
    // triggers to specific pools.
    // For usability,
    // Provide a way for triggers sharing/inheritance possibly with following modes
    // ONLY - only to pool
    // INHERIT - child pools inherit from parent
    // GLOBAL - all pools inherit
    if (plan != null && plan.isSetTriggers() && plan.isSetPoolTriggers()) {
      Map<String, Trigger> triggers = new HashMap<>();
      for (WMTrigger trigger : plan.getTriggers()) {
        ExecutionTrigger execTrigger = ExecutionTrigger.fromWMTrigger(trigger);
        triggers.put(trigger.getTriggerName(), execTrigger);
      }
      for (WMPoolTrigger poolTrigger : plan.getPoolTriggers()) {
        PoolState pool = pools.get(poolTrigger.getPool());
        Trigger trigger = triggers.get(poolTrigger.getTrigger());
        pool.triggers.add(trigger);
        poolsToRedistribute.add(pool.fullName);
        LOG.info("Adding pool " + pool.fullName + " trigger " + trigger);
      }
    }

    if (oldPools != null && !oldPools.isEmpty()) {
      // Looks like some pools were removed; kill running queries, re-queue the queued ones.
      for (PoolState oldPool : oldPools.values()) {
        oldPool.destroy(syncWork, e.getRequests, e.toReuse);
      }
    }

    LOG.info("Updating with " + totalQueryParallelism + " total query parallelism");
    int deltaSessions = totalQueryParallelism - this.totalQueryParallelism;
    this.totalQueryParallelism = totalQueryParallelism;
    if (deltaSessions == 0) return; // Nothing to do.
    if (deltaSessions < 0) {
      // First, see if we have sessions that we were planning to restart/kill; get rid of those.
      deltaSessions = transferSessionsToDestroy(
          syncWork.toKillQuery.keySet(), syncWork.toDestroyNoRestart, deltaSessions);
      deltaSessions = transferSessionsToDestroy(
          syncWork.toRestartInUse, syncWork.toDestroyNoRestart, deltaSessions);
    }
    if (deltaSessions != 0) {
      failOnFutureFailure(tezAmPool.resizeAsync(deltaSessions, syncWork.toDestroyNoRestart));
    }
  }
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