@Override
  public boolean canCommit(TaskAttemptId taskAttemptID) {
    readLock.lock();
    boolean canCommit = false;
    try {
      if (commitAttempt != null) {
        canCommit = taskAttemptID.equals(commitAttempt);
        LOG.info("Result of canCommit for " + taskAttemptID + ":" + canCommit);
      }
    } finally {
      readLock.unlock();
    }
    return canCommit;
  }
@Test
  public void testKillDuringTaskAttemptCommit() {
    mockTask = createMockTask(TaskType.REDUCE);        
    TaskId taskId = getNewTaskID();
    scheduleTaskAttempt(taskId);
    
    launchTaskAttempt(getLastAttempt().getAttemptId());
    updateLastAttemptState(TaskAttemptState.COMMIT_PENDING);
    commitTaskAttempt(getLastAttempt().getAttemptId());

    TaskAttemptId commitAttempt = getLastAttempt().getAttemptId();
    updateLastAttemptState(TaskAttemptState.KILLED);
    killRunningTaskAttempt(commitAttempt);

    assertFalse(mockTask.canCommit(commitAttempt));
  }