@Override
  public void dagComplete() {
    // This is effectively DAG completed, and can be used to reset statistics being tracked.
    LOG.info("DAG: " + dagCounter.get() + " completed. Scheduling stats: " + dagStats);
    dagCounter.incrementAndGet();
    if (metrics != null) {
      metrics.incrCompletedDagCount();
    }
    long tgVersionForZk;
    writeLock.lock();
    try {
      dagRunning = false;
      dagStats = new StatsPerDag();
      int pendingCount = 0;
      for (Entry<Priority, List<TaskInfo>> entry : pendingTasks.entrySet()) {
        if (entry.getValue() != null) {
          pendingCount += entry.getValue().size();
        }
      }
      int runningCount = 0;
      // We don't send messages to pending tasks with the flags; they should be killed elsewhere.
      for (Entry<Integer, TreeSet<TaskInfo>> entry : guaranteedTasks.entrySet()) {
        TreeSet<TaskInfo> set = speculativeTasks.get(entry.getKey());
        if (set == null) {
          set = new TreeSet<>();
          speculativeTasks.put(entry.getKey(), set);
        }
        for (TaskInfo info : entry.getValue()) {
          synchronized (info) {
            info.isGuaranteed = false;
          }
          set.add(info);
        }
      }
      guaranteedTasks.clear();
      for (Entry<Integer, TreeSet<TaskInfo>> entry : speculativeTasks.entrySet()) {
        if (entry.getValue() != null) {
          runningCount += entry.getValue().size();
        }
      }

      totalGuaranteed = unusedGuaranteed = 0;
      tgVersionForZk = ++totalGuaranteedVersion;
      if (metrics != null) {
        metrics.setDagId(null);
        // We remove the tasks above without state checks so just reset all metrics to 0.
        metrics.resetWmMetrics();
      }
      LOG.info("DAG reset. Current knownTaskCount={}, pendingTaskCount={}, runningTaskCount={}",
          knownTasks.size(), pendingCount, runningCount);
    } finally {
      writeLock.unlock();
    }
    if (workloadManagementEnabled) {
      updateGuaranteedInRegistry(tgVersionForZk, 0);
    }
    // TODO Cleanup pending tasks etc, so that the next dag is not affected.
  }
The difference between this and getTotalResources() is that this only gives currently free
resource instances, while the other lists all the instances that may become available in a
while.
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java