private void updatePreemptionListAndNotify(EndReason reason) {
      // if this task was added to pre-emption list, remove it
      if (enablePreemption) {
        String state = reason == null ? "FAILED" : reason.name();
        boolean removed = removeFromPreemptionQueueUnlocked(taskWrapper);
        if (removed) {
          TaskRunnerCallable trc = taskWrapper.getTaskRunnerCallable();
          LOG.info(TaskRunnerCallable.getTaskIdentifierString(trc.getRequest(),
              trc.getVertexSpec(), trc.getQueryId()) + " request " + state + "! Removed from preemption list.");
        }
      }

      numSlotsAvailable.incrementAndGet();
      metrics.setNumExecutorsAvailable(numSlotsAvailable.get());
      if (LOG.isDebugEnabled()) {
        LOG.debug("Task {} complete. WaitQueueSize={}, numSlotsAvailable={}, preemptionQueueSize={}",
          taskWrapper.getRequestId(), waitQueue.size(), numSlotsAvailable.get(),
          preemptionQueue.size());
      }
      synchronized (lock) {
        if (!waitQueue.isEmpty()) {
          lock.notifyAll();
        }
      }
    }
Victim Task (A) should be preempted in favor of a candidate Task (B) when:
   1. A is NOT on the same Vertex as B AND
     1.1. B is a Guaranteed Task while A is not OR
     1.2. Both are guaranteed but A is not finishable and B is
To make sure that Victim task is not behind some upstream updates (asynchronous),
we check its sources' state (by QueryFragmentInfo.canFinish method)
@param candidate Task
@param victim Task
@return True when victim should be preempted in favor of candidate Task
llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorService.java