code: public boolean shouldClose(TaskAttemptID taskid) {
      /**
       * If the task hasn't been closed yet, and it belongs to a completed
       * TaskInProgress close it.
       * 
       * However, for completed map tasks we do not close the task which
       * actually was the one responsible for _completing_ the TaskInProgress. 
       */
  
      if (tasksReportedClosed.contains(taskid)) {
        if (tasksToKill.keySet().contains(taskid))
          return true;
        else
          return false;
      }
  
      boolean close = false;
      TaskStatus ts = taskStatuses.get(taskid);
  
      if ((ts != null) &&
          ((this.failed) ||
          ((job.getStatus().getRunState() != JobStatus.RUNNING &&
           (job.getStatus().getRunState() != JobStatus.PREP))))) {
        tasksReportedClosed.add(taskid);
        close = true;
      } else if ((completes > 0) && // isComplete() is synchronized!
                 !(isMapTask() && !jobSetup && 
                   !jobCleanup && isComplete(taskid))) {
        tasksReportedClosed.add(taskid);
        close = true; 
      } else if (isCommitPending(taskid) && !shouldCommit(taskid)) {
        tasksReportedClosed.add(taskid);
        close = true; 
      } else {
        close = tasksToKill.keySet().contains(taskid);
      }   
      return close;
    }
docs: Returns whether a component task-thread should be
  closed because the containing JobInProgress has completed
  or the task is killed by the user