private void updateFallOffStats(
        String requestId) {
      long now = clock.getTime();
      FragmentCompletion fragmentCompletion =
          completingFragmentMap.remove(requestId);
      if (fragmentCompletion == null) {
        LOG.warn(
            "Received onSuccess/onFailure for a fragment for which a completing message was not received: {}",
            requestId);
        // Happens due to AM side pre-emption, or the AM asking for a task to die.
        // There's no hooks at the moment to get information over.
        // For now - decrement the count to avoid accounting errors.
        runningFragmentCount.decrementAndGet();
        // TODO: Extend TaskRunner2 or see if an API with callbacks will work
      } else {
        long timeTaken = now - fragmentCompletion.completingTime;
        switch (fragmentCompletion.state) {
          case SUCCESS:
          metrics.addMetricsFallOffSuccessTimeLost(timeTaken);
            break;
          case FAILED:
          metrics.addMetricsFallOffFailedTimeLost(timeTaken);
            break;
          case KILLED:
          metrics.addMetricsFallOffKilledTimeLost(timeTaken);
            break;
        }
      }
    }
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