@VisibleForTesting
  protected void schedulePendingTasks() throws InterruptedException {
    // Early exit for scheduler calls that came **before** the end of this run and no resources are available.
    // Preemption and locality delay will be taken care of in the first run.
    if (isClusterCapacityFull.get()) {
      return;
    }
    Ref<TaskInfo> downgradedTask = new Ref<>(null);
    Pair<Resource, Map<String, List<NodeInfo>>> availabilityPair = getResourceAvailability();
    writeLock.lock();
    try {
      if (LOG.isDebugEnabled()) {
        LOG.debug("ScheduleRun: {}", constructPendingTaskCountsLogMessage());
      }
      Iterator<Entry<Priority, List<TaskInfo>>> pendingIterator = pendingTasks.entrySet().iterator();
      while (pendingIterator.hasNext() && shouldCycle(availabilityPair.getRight())) {
        Entry<Priority, List<TaskInfo>> entry = pendingIterator.next();
        List<TaskInfo> taskListAtPriority = entry.getValue();
        Iterator<TaskInfo> taskIter = taskListAtPriority.iterator();
        boolean scheduledAllAtPriority = true;
        while (taskIter.hasNext()) {
          TaskInfo taskInfo = taskIter.next();
          if (taskInfo.getNumPreviousAssignAttempts() == 1) {
            dagStats.registerDelayedAllocation();
          }
          taskInfo.triedAssigningTask();
          ScheduleResult scheduleResult = scheduleTask(taskInfo, availabilityPair, downgradedTask);
          // Note: we must handle downgradedTask after this. We do it at the end, outside the lock.
          LOG.debug("ScheduleResult for Task: {} = {}", taskInfo, scheduleResult);
          if (scheduleResult == ScheduleResult.SCHEDULED) {
            taskIter.remove();
          } else {
            if (scheduleResult == ScheduleResult.INADEQUATE_TOTAL_RESOURCES) {
              LOG.info("Inadequate total resources before scheduling pending tasks." +
                  " Signalling scheduler timeout monitor thread to start timer.");
              startTimeoutMonitor();
              // TODO Nothing else should be done for this task. Move on.
            }

            if (scheduleResult == ScheduleResult.DELAYED_LOCALITY) {
              // Add the task to the delayed task queue if it does not already exist.
              maybeAddToDelayedTaskQueue(taskInfo);
            }

            // Add to HighPriority queue to potentially preempt a lower priority task and unblock this one
            maybeAddToHighPriorityTaskQueue(taskInfo);

            // Since there was an allocation failure - don't try assigning tasks at the next priority.
            scheduledAllAtPriority = false;
            // Don't break if this allocation failure was a result of a LOCALITY_DELAY. Others could still be allocated.
            if (scheduleResult != ScheduleResult.DELAYED_LOCALITY) {
              break;
            }
          } // end of else - i.e. could not allocate
        } // end of loop over pending tasks
        if (taskListAtPriority.isEmpty()) {
          // Remove the entry, if there's nothing left at the specific priority level
          pendingIterator.remove();
        }
        if (!scheduledAllAtPriority) {
          LOG.debug(
              "Unable to schedule all requests at priority={}. Skipping subsequent priority levels",
              entry.getKey());
          // Don't attempt scheduling for additional priorities
          break;
        }
      }
      // Finally take care of preemption requests that can unblock higher-pri tasks.
      // This removes preemptable tasks from the runningList and sends out a preempt request to the system.
      // Subsequent tasks will be scheduled once the de-allocate request for the preempted task is processed.
      while (!preemptionCandidates.isEmpty()) {
        TaskInfo toPreempt = preemptionCandidates.take();
        // 1. task has not terminated
        if (toPreempt.isGuaranteed != null) {
          String host = toPreempt.getAssignedNode().getHost();
           // 2. is currently assigned 3. no preemption pending on that Host
          if (toPreempt.getState() == TaskInfo.State.ASSIGNED &&
              (pendingPreemptionsPerHost.get(host) == null || pendingPreemptionsPerHost.get(host).intValue() == 0)) {
            LOG.info("Preempting {} running at Host={}", toPreempt, host);
            dagStats.registerTaskPreempted(toPreempt.getAssignedNode().getHost());
            registerPendingPreemption(toPreempt.getAssignedNode().getHost());
            toPreempt.setPreemptedInfo(clock.getTime());
            // Task cleanup
            TreeMap<Integer, TreeSet<LlapTaskSchedulerService.TaskInfo>> taskMap =
                toPreempt.isGuaranteed ? guaranteedTasks : speculativeTasks;
            taskMap.get(toPreempt.priority.getPriority()).remove(toPreempt);
            if (taskMap.get(toPreempt.priority.getPriority()).isEmpty()) {
              taskMap.remove(toPreempt.priority.getPriority());
            }
            // Preemption will finally be registered as a deallocateTask as a result of preemptContainer
            // That resets preemption info and allows additional tasks to be preempted at that Host if required
            getContext().preemptContainer(toPreempt.containerId);
          } else {
            // Maybe consider for future preemptions
            toPreempt.setPreemptedTime(0L);
          }
        }
      }
    } finally {
      writeLock.unlock();
    }
    if (downgradedTask.value != null) {
      WM_LOG.info("Downgrading " + downgradedTask.value.attemptId);
      checkAndSendGuaranteedStateUpdate(downgradedTask.value);
    }
  }
Should cycle when:
 1. There are available Resources OR
 2. There is a pending task with higher pri than running ones
@param availableHostMap
@return
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java