private void processCurrentEvents(EventState e, WmThreadSyncWork syncWork) throws Exception {
    // The order of processing is as follows. We'd reclaim or kill all the sessions that we can
    // reclaim from various user actions and errors, then apply the new plan if any,
    // then give out all we can give out (restart, get and reopen callers) and rebalance the
    // resource allocations in all the affected pools.
    // For every session, we'd check all the concurrent things happening to it.

    // TODO: also account for Tez-internal session restarts;
    //       AM reg info changes; add notifications, ignore errors, and update alloc.
    HashSet<String> poolsToRedistribute = new HashSet<>();

    // 0. Handle initialization results.
    for (SessionInitContext sw : e.initResults) {
      handleInitResultOnMasterThread(sw, syncWork, poolsToRedistribute);
    }
    e.initResults.clear();

    // 1. Handle kill query results - part 1, just put them in place. We will resolve what
    //    to do with the sessions after we go thru all the concurrent user actions.
    for (Map.Entry<WmTezSession, Boolean> entry : e.killQueryResults.entrySet()) {
      WmTezSession killQuerySession = entry.getKey();
      boolean killResult = entry.getValue();
      LOG.debug("Processing KillQuery {} for {}",
          killResult ? "success" : "failure", killQuerySession);
      // Note: do not cancel any user actions here; user actions actually interact with kills.
      KillQueryContext killCtx = killQueryInProgress.get(killQuerySession);
      if (killCtx == null) {
        LOG.error("Internal error - cannot find the context for killing {}", killQuerySession);
        continue;
      }
      killCtx.handleKillQueryCallback(!killResult);
    }
    e.killQueryResults.clear();

    // 2. Handle sessions that are being destroyed by users. Destroy implies return.
    for (WmTezSession sessionToDestroy : e.toDestroy) {
      if (e.toReturn.remove(sessionToDestroy)) {
        LOG.warn("The session was both destroyed and returned by the user; destroying");
      }
      LOG.info("Destroying {}", sessionToDestroy);
      RemoveSessionResult rr = handleReturnedInUseSessionOnMasterThread(
          e, sessionToDestroy, poolsToRedistribute, false);
      if (rr == RemoveSessionResult.OK || rr == RemoveSessionResult.NOT_FOUND) {
        // Restart even if there's an internal error.
        syncWork.toRestartInUse.add(sessionToDestroy);
      }
    }
    e.toDestroy.clear();

    // 3. Now handle actual returns. Sessions may be returned to the pool or may trigger expires.
    for (WmTezSession sessionToReturn: e.toReturn) {
      LOG.info("Returning {}", sessionToReturn);
      RemoveSessionResult rr = handleReturnedInUseSessionOnMasterThread(
          e, sessionToReturn, poolsToRedistribute, true);
      switch (rr) {
      case OK:
        WmEvent wmEvent = new WmEvent(WmEvent.EventType.RETURN);
        boolean wasReturned = tezAmPool.returnSessionAsync(sessionToReturn);
        if (!wasReturned) {
          syncWork.toDestroyNoRestart.add(sessionToReturn);
        } else {
          if (sessionToReturn.getWmContext() != null && sessionToReturn.getWmContext().isQueryCompleted()) {
            sessionToReturn.resolveReturnFuture();
          }
          wmEvent.endEvent(sessionToReturn);
        }
        break;
      case NOT_FOUND:
        syncWork.toRestartInUse.add(sessionToReturn); // Restart if there's an internal error.
        break;
      case IGNORE:
        break;
      default: throw new AssertionError("Unknown state " + rr);
      }
    }
    e.toReturn.clear();

    // 4. Reopen is essentially just destroy + get a new session for a session in use.
    for (Map.Entry<WmTezSession, SettableFuture<WmTezSession>> entry : e.toReopen.entrySet()) {
      LOG.info("Reopening {}", entry.getKey());
      handeReopenRequestOnMasterThread(
        e, entry.getKey(), entry.getValue(), poolsToRedistribute, syncWork);
    }
    e.toReopen.clear();

    // 5. All the sessions in use that were not destroyed or returned with a failed update now die.
    for (Map.Entry<WmTezSession, Integer> entry : e.updateErrors.entrySet()) {
      WmTezSession sessionWithUpdateError = entry.getKey();
      int failedEndpointVersion = entry.getValue();
      LOG.info("Update failed for {}", sessionWithUpdateError);
      handleUpdateErrorOnMasterThread(
          sessionWithUpdateError, failedEndpointVersion, e.toReuse, syncWork, poolsToRedistribute);
    }
    e.updateErrors.clear();

    // 6. Now apply a resource plan if any. This is expected to be pretty rare.
    boolean hasRequeues = false;
    if (e.resourcePlanToApply != null || e.doClearResourcePlan) {
      LOG.info("Applying new resource plan");
      int getReqCount = e.getRequests.size();
      applyNewResourcePlanOnMasterThread(e, syncWork, poolsToRedistribute);
      hasRequeues = getReqCount != e.getRequests.size();
    }
    e.resourcePlanToApply = null;
    e.doClearResourcePlan = false;

    // 7. Handle any move session requests. The way move session works right now is
    // a) sessions get moved to destination pool if there is capacity in destination pool
    // b) if there is no capacity in destination pool, the session gets killed (since we cannot pause a query)
    // TODO: in future this the process of killing can be delayed until the point where a session is actually required.
    // We could consider delaying the move (when destination capacity is full) until there is claim in src pool.
    // May be change command to support ... DELAYED MOVE TO etl ... which will run under src cluster fraction as long
    // as possible
    Map<WmTezSession, WmEvent> recordMoveEvents = new HashMap<>();
    boolean convertToDelayedMove = HiveConf.getBoolVar(conf, ConfVars.HIVE_SERVER2_WM_DELAYED_MOVE);
    for (MoveSession moveSession : e.moveSessions) {
      handleMoveSessionOnMasterThread(moveSession, syncWork, poolsToRedistribute, e.toReuse,
          recordMoveEvents, convertToDelayedMove);
    }
    e.moveSessions.clear();

    // 8. Handle all the get/reuse requests. We won't actually give out anything here, but merely
    //    map all the requests and place them in an appropriate order in pool queues. The only
    //    exception is the reuse without queue contention; can be granted immediately. If we can't
    //    reuse the session immediately, we will convert the reuse to a normal get, because we
    //    want query level fairness, and don't want the get in queue to hold up a session.
    GetRequest req;
    while ((req = e.getRequests.pollFirst()) != null) {
      LOG.info("Processing a new get request from " + req.mappingInput);
      queueGetRequestOnMasterThread(req, poolsToRedistribute, syncWork);
    }
    e.toReuse.clear();

    // 9. If delayed move is set to true, process the "delayed moves" now.
    if (HiveConf.getBoolVar(conf, ConfVars.HIVE_SERVER2_WM_DELAYED_MOVE)) {
      for (String poolName : pools.keySet()) {
        processDelayedMovesForPool(poolName, poolsToRedistribute, recordMoveEvents, syncWork, e.toReuse);
      }
    }

    // 10. Resolve all the kill query requests in flight. Nothing below can affect them.
    Iterator<KillQueryContext> iter = killQueryInProgress.values().iterator();
    while (iter.hasNext()) {
      KillQueryContext ctx = iter.next();
      KillQueryResult kr = ctx.process();
      switch (kr) {
      case IN_PROGRESS: continue; // Either the user or the kill is not done yet.
      case OK: {
        iter.remove();
        LOG.debug("Kill query succeeded; returning to the pool: {}", ctx.session);
        ctx.killSessionFuture.set(true);
        WmEvent wmEvent = new WmEvent(WmEvent.EventType.RETURN);
        if (!tezAmPool.returnSessionAsync(ctx.session)) {
          syncWork.toDestroyNoRestart.add(ctx.session);
        } else {
          if (ctx.session.getWmContext() != null && ctx.session.getWmContext().isQueryCompleted()) {
            ctx.session.resolveReturnFuture();
          }
          wmEvent.endEvent(ctx.session);
        }

        // Running query metrics needs to be updated for the pool
        updatePoolMetricsAfterKillTrigger(poolsToRedistribute, ctx);
        break;
      }
      case RESTART_REQUIRED: {
        iter.remove();
        ctx.killSessionFuture.set(true);
        LOG.debug("Kill query failed; restarting: {}", ctx.session);
        // Note: we assume here the session, before we resolve killQuery result here, is still
        //       "in use". That is because all the user ops above like return, reopen, etc.
        //       don't actually return/reopen/... when kill query is in progress.
        syncWork.toRestartInUse.add(ctx.session);

        // Running query metrics needs to be updated for the pool
        updatePoolMetricsAfterKillTrigger(poolsToRedistribute, ctx);
        break;
      }
      default: throw new AssertionError("Unknown state " + kr);
      }
    }

    // 11. If there was a cluster state change, make sure we redistribute all the pools.
    if (e.hasClusterStateChanged) {
      LOG.info("Processing a cluster state change");
      poolsToRedistribute.addAll(pools.keySet());
      e.hasClusterStateChanged = false;
    }

    // 12. Finally, for all the pools that have changes, promote queued queries and rebalance.
    for (String poolName : poolsToRedistribute) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Processing changes for pool " + poolName + ": " + pools.get(poolName));
      }
      processPoolChangesOnMasterThread(poolName, hasRequeues, syncWork);
    }

    // 13. Save state for future iterations.
    for (KillQueryContext killCtx : syncWork.toKillQuery.values()) {
      if (killQueryInProgress.put(killCtx.session, killCtx) != null) {
        LOG.error("One query killed several times - internal error {}", killCtx.session);
      }
    }

    // 14. To record move events, we need to cluster fraction updates that happens at step 12.
    for (Map.Entry<WmTezSession, WmEvent> entry : recordMoveEvents.entrySet()) {
      entry.getValue().endEvent(entry.getKey());
    }

    // 15. Give our final state to UI/API requests if any.
    if (e.dumpStateFuture != null) {
      List<String> result = new ArrayList<>();
      result.add("RESOURCE PLAN " + rpName + "; default pool " + defaultPool);
      for (PoolState ps : pools.values()) {
        dumpPoolState(ps, result);
      }
      e.dumpStateFuture.set(result);
      e.dumpStateFuture = null;
    }

    // 16. Notify tests and global async ops.
    for (SettableFuture<Boolean> testEvent : e.testEvents) {
      LOG.info("Triggering test event " + System.identityHashCode(testEvent));
      testEvent.set(null);
    }
    e.testEvents.clear();

    if (e.applyRpFuture != null) {
      e.applyRpFuture.set(true);
      e.applyRpFuture = null;
    }
  }
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