@Override
  public int execute() {
    int rc = 1;
    boolean cleanContext = false;
    Context ctx = null;
    Ref<TezSessionState> sessionRef = Ref.from(null);

    final String queryId = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_QUERY_ID);

    TezJobMonitor monitor = null;
    try {
      // Get or create Context object. If we create it we have to clean it later as well.
      ctx = context;
      if (ctx == null) {
        ctx = new Context(conf);
        cleanContext = true;
        // some DDL task that directly executes a TezTask does not setup Context and hence TriggerContext.
        // Setting queryId is messed up. Some DDL tasks have executionId instead of proper queryId.
        WmContext wmContext = new WmContext(System.currentTimeMillis(), queryId);
        ctx.setWmContext(wmContext);
      }
      // Need to remove this static hack. But this is the way currently to get a session.
      SessionState ss = SessionState.get();
      // Note: given that we return pool sessions to the pool in the finally block below, and that
      //       we need to set the global to null to do that, this "reuse" may be pointless.
      TezSessionState session = sessionRef.value = ss.getTezSession();
      if (session != null && !session.isOpen()) {
        LOG.warn("The session: " + session + " has not been opened");
      }

      // We only need a username for UGI to use for groups; getGroups will fetch the groups
      // based on Hadoop configuration, as documented at
      // https://hadoop.apache.org/docs/r2.8.0/hadoop-project-dist/hadoop-common/GroupsMapping.html
      String userName = getUserNameForGroups(ss);
      List<String> groups = null;
      if (userName == null) {
        userName = "anonymous";
      } else {
        try {
          groups = UserGroupInformation.createRemoteUser(userName).getGroups();
        } catch (Exception ex) {
          LOG.warn("Cannot obtain groups for " + userName, ex);
        }
      }
      MappingInput mi = new MappingInput(userName, groups,
          ss.getHiveVariables().get("wmpool"), ss.getHiveVariables().get("wmapp"));

      WmContext wmContext = ctx.getWmContext();
      String executionMode = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_EXECUTION_MODE);
      runtimeContext.setExecutionMode(executionMode);
      // jobConf will hold all the configuration for hadoop, tez, and hive, which are not set in AM defaults
      JobConf jobConf = utils.createConfiguration(conf, false);

      // Setup the job specific keystore path if exists and put the password into the environment variables of tez am/tasks.
      HiveConfUtil.updateJobCredentialProviders(jobConf);

      // Get all user jars from work (e.g. input format stuff).
      String[] allNonConfFiles = work.configureJobConfAndExtractJars(jobConf);
      // DAG scratch dir. We get a session from the pool so it may be different from Tez one.
      // TODO: we could perhaps reuse the same directory for HiveResources?
      Path scratchDir = utils.createTezDir(ctx.getMRScratchDir(), conf);
      CallerContext callerContext =
          CallerContext.create("HIVE", String.format(USER_ID, queryPlan.getQueryId(), userName), "HIVE_QUERY_ID",
              queryPlan.getQueryStr());

      perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.TEZ_GET_SESSION);
      session = sessionRef.value = WorkloadManagerFederation.getSession(
          sessionRef.value, conf, mi, getWork().getLlapMode(), wmContext);
      perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.TEZ_GET_SESSION);

      try {
        ss.setTezSession(session);
        LOG.info("Subscribed to counters: {} for queryId: {}", wmContext.getSubscribedCounters(),
            wmContext.getQueryId());

        // Ensure the session is open and has the necessary local resources.
        // This would refresh any conf resources and also local resources.
        ensureSessionHasResources(session, allNonConfFiles);

        // This is a combination of the jar stuff from conf, and not from conf.
        List<LocalResource> allNonAppResources = session.getLocalizedResources();
        logResources(allNonAppResources);

        Map<String, LocalResource> allResources = DagUtils.createTezLrMap(
            session.getAppJarLr(), allNonAppResources);

        // next we translate the TezWork to a Tez DAG
        DAG dag = build(jobConf, work, scratchDir, ctx, allResources);
        dag.setCallerContext(callerContext);

        // Note: we no longer call addTaskLocalFiles because all the resources are correctly
        //       updated in the session resource lists now, and thus added to vertices.
        //       If something breaks, dag.addTaskLocalFiles might need to be called here.

        // Check isShutdown opportunistically; it's never unset.
        if (this.isShutdown) {
          throw new HiveException("Operation cancelled");
        }
        DAGClient dagClient = submit(dag, sessionRef);
        session = sessionRef.value;
        boolean wasShutdown = false;
        synchronized (dagClientLock) {
          assert this.dagClient == null;
          wasShutdown = this.isShutdown;
          if (!wasShutdown) {
            this.dagClient = dagClient;
          }
        }
        if (wasShutdown) {
          closeDagClientOnCancellation(dagClient);
          throw new HiveException("Operation cancelled");
        }

        // Log all the info required to find the various logs for this query
        String dagId = this.dagClient.getDagIdentifierString();
        String appId = this.dagClient.getSessionIdentifierString();
        LOG.info("HS2 Host: [{}], Query ID: [{}], Dag ID: [{}], DAG App ID: [{}], DAG App address: [{}]",
            ServerUtils.hostname(), queryId, dagId, appId, session.getSession().getAmHost());
        LogUtils.putToMDC(LogUtils.DAGID_KEY, dagId);
        this.jobID = dagId;
        runtimeContext.setDagId(dagId);
        runtimeContext.setSessionId(session.getSessionId());
        runtimeContext.setApplicationId(appId);

        // finally monitor will print progress until the job is done
        monitor = new TezJobMonitor(session, work.getAllWork(), dagClient, conf, dag, ctx, runtimeContext.counters,
            perfLogger);
        runtimeContext.setMonitor(monitor);
        rc = monitor.monitorExecution();

        if (rc != 0) {
          this.setException(new TezRuntimeException(dagId, monitor.getDiagnostics()));
        }

        try {
          // fetch the counters
          Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
          DAGStatus dagStatus = dagClient.getDAGStatus(statusGetOpts);
          this.setStatusMessage(dagStatus.getState().name());

          TezCounters dagCounters = dagStatus.getDAGCounters();

          // if initial counters exists, merge it with dag counters to get aggregated view
          TezCounters mergedCounters = runtimeContext.counters == null ? dagCounters : Utils.mergeTezCounters(
              dagCounters, runtimeContext.counters);
          runtimeContext.counters = mergedCounters;
        } catch (Exception err) {
          // Don't fail execution due to counters - just don't print summary info
          LOG.warn("Failed to get counters. Ignoring, summary info will be incomplete.", err);
          runtimeContext.counters = null;
        }

        // save useful commit information into query state, e.g. for custom commit hooks, like Iceberg
        if (rc == 0) {
          collectCommitInformation(work);
        }
      } finally {
        // Note: due to TEZ-3846, the session may actually be invalid in case of some errors.
        //       Currently, reopen on an attempted reuse will take care of that; we cannot tell
        //       if the session is usable until we try.
        // We return this to the pool even if it's unusable; reopen is supposed to handle this.
        wmContext = ctx.getWmContext();
        try {
          if (sessionRef.value != null) {
            sessionRef.value.returnToSessionManager();
          }
        } catch (Exception e) {
          LOG.error("Failed to return session: {} to pool", session, e);
          throw e;
        }

        if (!conf.getVar(HiveConf.ConfVars.TEZ_SESSION_EVENTS_SUMMARY).equalsIgnoreCase("none") &&
            wmContext != null) {
          if (conf.getVar(HiveConf.ConfVars.TEZ_SESSION_EVENTS_SUMMARY).equalsIgnoreCase("json")) {
            wmContext.printJson(console);
          } else if (conf.getVar(HiveConf.ConfVars.TEZ_SESSION_EVENTS_SUMMARY).equalsIgnoreCase("text")) {
            wmContext.print(console);
          }
        }
      }

      if (LOG.isInfoEnabled() && runtimeContext.counters != null
          && (HiveConf.getBoolVar(conf, HiveConf.ConfVars.TEZ_EXEC_SUMMARY) ||
          Utilities.isPerfOrAboveLogging(conf))) {
        for (CounterGroup group : runtimeContext.counters) {
          monitor.logger().printInfo(group.getDisplayName() + ":");
          for (TezCounter counter : group) {
            monitor.logger().printInfo("   " + counter.getDisplayName() + ": " + counter.getValue());
          }
        }
      }
      updateNumRows();
    } catch (Exception e) {
      LOG.error("Failed to execute tez graph.", e);
      setException(e);
      // rc will be 1 at this point indicating failure.
    } finally {
      Utilities.clearWork(conf);

      // Clear gWorkMap
      for (BaseWork w : work.getAllWork()) {
        JobConf workCfg = workToConf.get(w);
        if (workCfg != null) {
          Utilities.clearWorkMapForConf(workCfg);
        }
      }

      if (cleanContext) {
        try {
          ctx.clear();
        } catch (Exception e) {
          /*best effort*/
          LOG.warn("Failed to clean up after tez job", e);
        }
      }
      // need to either move tmp files or remove them
      DAGClient dagClient = null;
      synchronized (dagClientLock) {
        dagClient = this.dagClient;
        this.dagClient = null;
      }
      // TODO: not clear why we don't do the rest of the cleanup if dagClient is not created.
      //       E.g. jobClose will be called if we fail after dagClient creation but no before...
      //       DagClient as such should have no bearing on jobClose.
      if (dagClient != null) {
        // rc will only be overwritten if close errors out
        rc = close(work, rc, dagClient);
      }
      if (monitor != null){
        monitor.logger().endSummary();
      }
    }
    return rc;
  }
Making TezTask backward compatible with the old MR-based Task API (ExecDriver/MapRedTask)
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java