public int monitorExecution() {
    boolean done = false;
    boolean success = false;
    int failedCounter = 0;
    final StopWatch failureTimer = new StopWatch();
    int rc = 0;
    DAGStatus status = null;
    Map<String, Progress> vertexProgressMap = null;


    long monitorStartTime = System.currentTimeMillis();
    synchronized (shutdownList) {
      shutdownList.add(dagClient);
    }
    perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.TEZ_RUN_DAG);
    perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.TEZ_SUBMIT_TO_RUNNING);
    DAGStatus.State lastState = null;
    boolean running = false;

    long checkInterval = HiveConf.getTimeVar(hiveConf, HiveConf.ConfVars.TEZ_DAG_STATUS_CHECK_INTERVAL,
      TimeUnit.MILLISECONDS);
    WmContext wmContext = null;
    while (true) {

      try {
        if (context != null) {
          context.checkHeartbeaterLockException();
        }

        wmContext = context.getWmContext();
        EnumSet<StatusGetOpts> opts = null;
        if (wmContext != null) {
          Set<String> desiredCounters = wmContext.getSubscribedCounters();
          if (desiredCounters != null && !desiredCounters.isEmpty()) {
            opts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
          }
        }

        status = dagClient.getDAGStatus(opts, checkInterval);
        session.updateDagStatus(status);

        vertexProgressMap = status.getVertexProgress();
        List<String> vertexNames = vertexProgressMap.keySet()
          .stream()
          .map(k -> k.replaceAll(" ", "_"))
          .collect(Collectors.toList());
        if (wmContext != null) {
          Set<String> desiredCounters = wmContext.getSubscribedCounters();
          TezCounters dagCounters = status.getDAGCounters();
          // if initial counters exists, merge it with dag counters to get aggregated view
          TezCounters mergedCounters = counters == null ? dagCounters : Utils.mergeTezCounters(dagCounters, counters);
          if (mergedCounters != null && desiredCounters != null && !desiredCounters.isEmpty()) {
            Map<String, Long> currentCounters = getCounterValues(mergedCounters, vertexNames, vertexProgressMap,
              desiredCounters, done);
            LOG.debug("Requested DAG status. checkInterval: {}. currentCounters: {}", checkInterval, currentCounters);
            wmContext.setCurrentCounters(currentCounters);
          }
        }
        DAGStatus.State state = status.getState();

        failedCounter = 0; // AM is responsive again (recovery?)
        failureTimer.reset();

        if (state != lastState || state == RUNNING) {
          lastState = state;

          switch (state) {
            case SUBMITTED:
              console.printInfo("Status: Submitted");
              break;
            case INITING:
              console.printInfo("Status: Initializing");
              this.executionStartTime = System.currentTimeMillis();
              break;
            case RUNNING:
              if (!running) {
                perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.TEZ_SUBMIT_TO_RUNNING);
                console.printInfo("Status: Running (" + dagClient.getExecutionContext() + ")\n");
                this.executionStartTime = System.currentTimeMillis();
                running = true;
              }
              updateFunction.update(status, vertexProgressMap);
              break;
            case SUCCEEDED:
              if (!running) {
                this.executionStartTime = monitorStartTime;
              }
              updateFunction.update(status, vertexProgressMap);
              success = true;
              running = false;
              done = true;
              break;
            case KILLED:
              if (!running) {
                this.executionStartTime = monitorStartTime;
              }
              updateFunction.update(status, vertexProgressMap);
              console.printInfo("Status: Killed");
              running = false;
              done = true;
              rc = 1;
              break;
            case FAILED:
            case ERROR:
              if (!running) {
                this.executionStartTime = monitorStartTime;
              }
              updateFunction.update(status, vertexProgressMap);
              console.printError("Status: Failed");
              running = false;
              done = true;
              rc = 2;
              break;
          }
        }
        if (wmContext != null && done) {
          wmContext.setQueryCompleted(true);
        }
      } catch (Exception e) {
        console.printInfo("Exception: " + e.getMessage());
        boolean isInterrupted = hasInterruptedException(e);
        if (failedCounter == 0) {
          failureTimer.reset();
          failureTimer.start();
        }
        if (isInterrupted
            || (++failedCounter >= MAX_RETRY_FAILURES && failureTimer.now(TimeUnit.MILLISECONDS) > MAX_RETRY_INTERVAL)) {
          try {
            if (isInterrupted) {
              console.printInfo("Killing DAG...");
            } else {
              console.printInfo(String.format("Killing DAG... after %d seconds",
                  failureTimer.now(TimeUnit.SECONDS)));
            }
            dagClient.tryKillDAG();
          } catch (IOException | TezException tezException) {
            // best effort
          }
          String reportedException = "Execution has failed, stack trace: " + ExceptionUtils.getStackTrace(e);
          console.printError(reportedException);
          diagnostics.append(reportedException);
          rc = 1;
          done = true;
        } else {
          console.printInfo("Retrying...");
        }
        if (wmContext != null && done) {
          wmContext.setQueryCompleted(true);
        }
      } finally {
        if (done) {
          if (wmContext != null && done) {
            wmContext.setQueryCompleted(true);
          }
          if (rc != 0 && status != null) {
            for (String diag : status.getDiagnostics()) {
              console.printError(diag);
              diagnostics.append(diag);
            }
          }
          synchronized (shutdownList) {
            shutdownList.remove(dagClient);
          }
          break;
        }
      }
    }

    perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.TEZ_RUN_DAG);
    printSummary(success, vertexProgressMap);
    return rc;
  }
TezJobMonitor keeps track of a tez job while it's being executed. It will
print status to the console and retrieve final status of the job after
completion.
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezJobMonitor.java