### Question:
TaskImpl implements Task, EventHandler<TaskEvent> { @Override public void handle(TaskEvent event) { if (LOG.isDebugEnabled()) { LOG.debug("Processing " + event.getTaskID() + " of type " + event.getType()); } try { writeLock.lock(); TaskStateInternal oldState = getInternalState(); try { stateMachine.doTransition(event.getType(), event); } catch (InvalidStateTransitonException e) { LOG.error("Can't handle this event at current state for " + this.taskId, e); internalError(event.getType()); } if (oldState != getInternalState()) { LOG.info(taskId + " Task Transitioned from " + oldState + " to " + getInternalState()); } } finally { writeLock.unlock(); } } TaskImpl(JobId jobId, TaskType taskType, int partition,
      EventHandler eventHandler, Path remoteJobConfFile, JobConf conf,
      TaskAttemptListener taskAttemptListener,
      Token<JobTokenIdentifier> jobToken,
      Credentials credentials, Clock clock,
      Map<TaskId, TaskInfo> completedTasksFromPreviousRun, int startCount,
      MRAppMetrics metrics, AppContext appContext); @Override TaskState getState(); @Override Map<TaskAttemptId, TaskAttempt> getAttempts(); @Override TaskAttempt getAttempt(TaskAttemptId attemptID); @Override TaskId getID(); @Override boolean isFinished(); @Override TaskReport getReport(); @Override Counters getCounters(); @Override float getProgress(); @VisibleForTesting TaskStateInternal getInternalState(); @Override boolean canCommit(TaskAttemptId taskAttemptID); @Override void handle(TaskEvent event);  }### Answer:
@Test public void testKillSuccessfulTask() { LOG.info("--- START: testKillSuccesfulTask ---"); mockTask = createMockTask(TaskType.MAP); TaskId taskId = getNewTaskID(); scheduleTaskAttempt(taskId); launchTaskAttempt(getLastAttempt().getAttemptId()); commitTaskAttempt(getLastAttempt().getAttemptId()); mockTask.handle(new TaskTAttemptEvent(getLastAttempt().getAttemptId(), TaskEventType.T_ATTEMPT_SUCCEEDED)); assertTaskSucceededState(); mockTask.handle(new TaskEvent(taskId, TaskEventType.T_KILL)); assertTaskSucceededState(); }
@Test public void testSpeculativeMapFetchFailure() { mockTask = createMockTask(TaskType.MAP); runSpeculativeTaskAttemptSucceeds(TaskEventType.T_ATTEMPT_KILLED); assertEquals(2, taskAttempts.size()); mockTask.handle(new TaskTAttemptEvent(taskAttempts.get(1).getAttemptId(), TaskEventType.T_ATTEMPT_FAILED)); assertTaskScheduledState(); assertEquals(3, taskAttempts.size()); }
@Test public void testSpeculativeMapMultipleSucceedFetchFailure() { mockTask = createMockTask(TaskType.MAP); runSpeculativeTaskAttemptSucceeds(TaskEventType.T_ATTEMPT_SUCCEEDED); assertEquals(2, taskAttempts.size()); mockTask.handle(new TaskTAttemptEvent(taskAttempts.get(1).getAttemptId(), TaskEventType.T_ATTEMPT_FAILED)); assertTaskScheduledState(); assertEquals(3, taskAttempts.size()); }
@Test public void testSpeculativeMapFailedFetchFailure() { mockTask = createMockTask(TaskType.MAP); runSpeculativeTaskAttemptSucceeds(TaskEventType.T_ATTEMPT_FAILED); assertEquals(2, taskAttempts.size()); mockTask.handle(new TaskTAttemptEvent(taskAttempts.get(1).getAttemptId(), TaskEventType.T_ATTEMPT_FAILED)); assertTaskScheduledState(); assertEquals(3, taskAttempts.size()); }