private void handeReopenRequestOnMasterThread(EventState e, WmTezSession session,
      SettableFuture<WmTezSession> future, HashSet<String> poolsToRedistribute,
      WmThreadSyncWork syncWork) throws Exception {
    if (e.updateErrors.remove(session) != null) {
      LOG.info("Ignoring an update error for a session being reopened");
    }
    GetRequest reuseRequest = e.toReuse.remove(session);
    if (reuseRequest != null) {
      reuseRequest.future.setException(new AssertionError("Invalid reuse attempt"));
    }
    // In order to expedite things in a general case, we are not actually going to reopen
    // anything. Instead, we will try to give out an existing session from the pool, and restart
    // the problematic one in background.
    String poolName = session.getPoolName();
    // Do not update metrics, we'd immediately add the session back if we are able to remove.
    RemoveSessionResult rr = checkAndRemoveSessionFromItsPool(
        session, poolsToRedistribute, false, false);
    switch (rr) {
    case OK:
      // If pool didn't exist, checkAndRemoveSessionFromItsPool wouldn't have returned OK.
      PoolState pool = pools.get(poolName);
      SessionInitContext sw = new SessionInitContext(future, poolName, session.getQueryId(),
          session.getWmContext(), session.extractHiveResources());
      // We have just removed the session from the same pool, so don't check concurrency here.
      pool.initializingSessions.add(sw);
      // Do not update metrics - see above.
      sw.start();
      syncWork.toRestartInUse.add(session);
      return;
    case IGNORE:
      // Reopen implies the use of the reopened session for the same query that we gave it out
      // for; so, as we would have failed an active query, fail the user before it's started.
      future.setException(new RuntimeException("WM killed this session during reopen: "
          + session.getReasonForKill()));
      return; // No longer relevant for WM.
    case NOT_FOUND:
      // If we fail to remove, it's probably an internal error. We'd try to handle it the same way
      // as above - by restarting the session. We'd fail the caller to avoid exceeding parallelism.
      future.setException(new RuntimeException("Reopen failed due to an internal error"));
      syncWork.toRestartInUse.add(session);
      return;
    default: throw new AssertionError("Unknown state " + rr);
    }
  }
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