@Override
    public void run() {
      try {
        Long lastKillTimeMs = null;
        SanityChecker sc = null;
        while (!isShutdown.get()) {
          RejectedExecutionException rejectedException = null;
          if (nextSanityCheck != null && ((nextSanityCheck - System.nanoTime()) <= 0)) {
            sc = sanityCheckQueue(sc);
            nextSanityCheck = null;
          }
          synchronized (lock) {
            // Since schedule() can be called from multiple threads, we peek the wait queue, try
            // scheduling the task and then remove the task if scheduling is successful. This
            // will make sure the task's place in the wait queue is held until it gets scheduled.
            task = waitQueue.peek();
            if (task == null) {
              waitOnLock();
              continue;
            }
            // If the task cannot finish and if no slots are available then don't schedule it.
            // Also don't wait if we have a task and we just killed something to schedule it.
            // (numSlotsAvailable can go negative, if the callback after the thread completes is delayed)
            boolean shouldWait = numSlotsAvailable.get() <= 0 && lastKillTimeMs == null;
            boolean canKill = false;
            if (task.canFinishForPriority() || task.isGuaranteed()) {
              if (LOG.isDebugEnabled()) {
                LOG.debug("Attempting to schedule task {}, canFinish={}. Current state: "
                    + "preemptionQueueSize={}, numSlotsAvailable={}, waitQueueSize={}",
                    task.getRequestId(), task.getTaskRunnerCallable().canFinish(),
                    preemptionQueue.size(), numSlotsAvailable.get(), waitQueue.size());
              }
              canKill = enablePreemption && canPreempt(task, preemptionQueue.peek());
              shouldWait = shouldWait && !canKill;
            }
            if (shouldWait) {
              waitOnLock();
              // Another task at a higher priority may have come in during the wait. Lookup the
              // queue again to pick up the task at the highest priority.
              continue;
            }
            nextSanityCheck = null; // We are going to do something useful now.
            try {
              tryScheduleUnderLock(task);
              lastKillTimeMs = null; // We have filled the spot we may have killed for (if any).
            } catch (RejectedExecutionException e) {
              rejectedException = e;
            }
          } // synchronized (lock)

          // Handle the rejection outside of the lock
          if (rejectedException != null) {
            if (lastKillTimeMs != null
                && (clock.getTime() - lastKillTimeMs) < PREEMPTION_KILL_GRACE_MS) {
              // We killed something, but still got rejected. Wait a bit to give a chance to our
              // previous victim to actually die.
              synchronized (lock) {
                lock.wait(PREEMPTION_KILL_GRACE_SLEEP_MS);
              }
            } else {
              if (LOG.isDebugEnabled() && lastKillTimeMs != null) {
                LOG.debug("Grace period ended for the previous kill; preemtping more tasks");
              }
              if (handleScheduleAttemptedRejection(task)) {
                lastKillTimeMs = clock.getTime(); // We killed something.
              }
            }
          }
        }
      } catch (InterruptedException e) {
        if (isShutdown.get()) {
          LOG.info(WAIT_QUEUE_SCHEDULER_THREAD_NAME_FORMAT
              + " thread has been interrupted after shutdown.");
        } else {
          LOG.warn(WAIT_QUEUE_SCHEDULER_THREAD_NAME_FORMAT + " interrupted without shutdown", e);
          throw new RuntimeException(e);
        }
      }
    }
Worker that takes tasks from wait queue and schedule it for execution.
llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java