@Override
  public SubmissionState schedule(TaskRunnerCallable task) {
    TaskWrapper taskWrapper = new TaskWrapper(task, this);
    SubmissionState result;
    TaskWrapper evictedTask;
    boolean canFinish;
    synchronized (lock) {
      // If the queue does not have capacity, it does not throw a Rejection. Instead it will
      // return the task with the lowest priority, which could be the task which is currently being processed.

      // TODO HIVE-11687 It's possible for a bunch of tasks to come in around the same time, without the
      // actual executor threads picking up any work. This will lead to unnecessary rejection of tasks.
      // The wait queue should be able to fit at least (waitQueue + currentFreeExecutor slots)
      if (LOG.isDebugEnabled()) {
        LOG.debug(
            "Offering to wait queue with: waitQueueSize={}, numSlotsAvailable={}, runningFragmentCount={} ",
            waitQueue.size(), numSlotsAvailable.get(),
            runningFragmentCount.get());
      }

      canFinish = taskWrapper.getTaskRunnerCallable().canFinish();
      taskWrapper.updateCanFinishForPriority(canFinish); // Update the property before offering.
      evictedTask = waitQueue.offer(taskWrapper, maxParallelExecutors - runningFragmentCount.get());
      // Finishable state is checked on the task, via an explicit query to the TaskRunnerCallable

      // null evicted task means offer accepted
      // evictedTask is not equal taskWrapper means current task is accepted and it evicted
      // some other task
      if (evictedTask == null || !evictedTask.equals(taskWrapper)) {
        knownTasks.put(taskWrapper.getRequestId(), taskWrapper);
        taskWrapper.setIsInWaitQueue(true);
        task.setWmCountersQueued();
        if (LOG.isDebugEnabled()) {
          LOG.debug("{} added to wait queue. Current wait queue size={}", task.getRequestId(),
              waitQueue.size());
        }

        result = evictedTask == null ? SubmissionState.ACCEPTED : SubmissionState.EVICTED_OTHER;

        if (LOG.isDebugEnabled() && evictedTask != null) {
          LOG.debug("Eviction: {} {} {}", taskWrapper, result, evictedTask);
        }
      } else {
        if (LOG.isInfoEnabled()) {
          LOG.info(
              "wait queue full, size={}. numSlotsAvailable={}, runningFragmentCount={}. {} not added",
              waitQueue.size(), numSlotsAvailable.get(), runningFragmentCount.get(), task.getRequestId());
        }
        evictedTask.getTaskRunnerCallable().killTask();

        result = SubmissionState.REJECTED;

        if (LOG.isDebugEnabled()) {
          LOG.debug("{} is {} as wait queue is full", taskWrapper.getRequestId(), result);
        }
        metrics.incrTotalRejectedRequests();
        return result;
      }

      // Register for notifications inside the lock. Should avoid races with unregisterForNotifications
      // happens in a different Submission thread. i.e. Avoid register running for this task
      // after some other submission has evicted it.
      boolean stateChanged = !taskWrapper.maybeRegisterForFinishedStateNotifications(canFinish);
      if (stateChanged) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Finishable state of {} updated to {} during registration for state updates",
              taskWrapper.getRequestId(), !canFinish);
        }
        finishableStateUpdated(taskWrapper, !canFinish);
      }
    }

    // At this point, the task has been added into the queue. It may have caused an eviction for
    // some other task.

    // This registration has to be done after knownTasks has been populated.
    // Register for state change notifications so that the waitQueue can be re-ordered correctly
    // if the fragment moves in or out of the finishable state.

    LOG.debug("Wait Queue: {}", waitQueue);

    if (evictedTask != null) {
      LOG.info("{} evicted from wait queue in favor of {} because of lower priority", evictedTask.getRequestId(),
          task.getRequestId());
      if (LOG.isDebugEnabled()) { // detailed info about the decision
        FragmentRuntimeInfo evictedInfo =
            evictedTask.getTaskRunnerCallable().getFragmentRuntimeInfo();
        FragmentRuntimeInfo taskInfo = task.getFragmentRuntimeInfo();

        int knownPendingTasksForEvicted = evictedInfo.getNumSelfAndUpstreamTasks()
            - evictedInfo.getNumSelfAndUpstreamCompletedTasks();
        int knownPendingTasksForCurrent =
            taskInfo.getNumSelfAndUpstreamTasks() - taskInfo.getNumSelfAndUpstreamCompletedTasks();

        long firstAttemptStartTimeEvicted = evictedInfo.getFirstAttemptStartTime();
        long firstAttemptStartTimeCurrent = taskInfo.getFirstAttemptStartTime();

        LOG.debug(
            "{} (guaranteed: {}, canFinishForPriority: {}, withinDagPriority: {}, currentAttemptStartTime: {}, "
                + "firstAttemptStartTime: {}, knownPending: {}) evicted from wait queue"
                + "in favor of {} (guaranteed: {}, canFinishForPriority: {}, withinDagPriority: {},"
                + " currentAttemptStartTime: {}, firstAttemptStartTime: {}, knownPending: {})"
                + "because of lower priority",
            evictedTask.getRequestId(), evictedTask.isGuaranteed(), evictedTask.canFinishForPriority(),
            evictedInfo.getWithinDagPriority(), evictedInfo.getCurrentAttemptStartTime(),
            firstAttemptStartTimeEvicted, knownPendingTasksForEvicted, task.getRequestId(),
            taskWrapper.isGuaranteed(), taskWrapper.canFinishForPriority(), taskInfo.getWithinDagPriority(),
            taskInfo.getCurrentAttemptStartTime(), firstAttemptStartTimeCurrent,
            knownPendingTasksForCurrent);
      }
      try {
        knownTasks.remove(evictedTask.getRequestId());
        evictedTask.maybeUnregisterForFinishedStateNotifications();
        evictedTask.setIsInWaitQueue(false);
      } finally {
        // This is dealing with tasks from a different submission, and cause the kill
        // to go out before the previous submissions has completed. Handled in the AM
        evictedTask.getTaskRunnerCallable().killTask();
      }
      metrics.incrTotalEvictedFromWaitQueue();
    }
    synchronized (lock) {
      lock.notifyAll();
    }
    metrics.setExecutorNumQueuedRequests(waitQueue.size());

    return result;
  }
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