### Question:
TaskImpl implements Task, EventHandler<TaskEvent> { @Override public void handle(TaskEvent event) { if (LOG.isDebugEnabled()) { LOG.debug("Processing TaskEvent " + event.getTaskID() + " of type " + event.getType() + " while in state " + getInternalState() + ". Event: " + event); } 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(TezVertexID vertexId, int taskIndex,
      EventHandler eventHandler, Configuration conf,
      TaskAttemptListener taskAttemptListener,
      Clock clock, TaskHeartbeatHandler thh, AppContext appContext,
      boolean leafVertex, Resource resource,
      ContainerContext containerContext); @Override TaskState getState(); @Override Map<TezTaskAttemptID, TaskAttempt> getAttempts(); @Override TaskAttempt getAttempt(TezTaskAttemptID attemptID); @Override Vertex getVertex(); @Override TezTaskID getTaskId(); @Override boolean isFinished(); @Override TaskReport getReport(); @Override TezCounters getCounters(); @Override float getProgress(); @Override List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID,
      int fromEventId, int maxEvents); @Override List<String> getDiagnostics(); @Override TaskState restoreFromEvent(HistoryEvent historyEvent); @VisibleForTesting TaskStateInternal getInternalState(); @Override boolean canCommit(TezTaskAttemptID taskAttemptID); @Override boolean needsWaitAfterOutputConsumable(); @Override TezTaskAttemptID getOutputConsumableAttempt(); @Override TaskAttempt getSuccessfulAttempt(); @Override void handle(TaskEvent event);  }### Answer:
@SuppressWarnings("rawtypes") @Test public void testTaskSucceedAndRetroActiveFailure() { TezTaskID taskId = getNewTaskID(); scheduleTaskAttempt(taskId); launchTaskAttempt(mockTask.getLastAttempt().getID()); updateAttemptState(mockTask.getLastAttempt(), TaskAttemptState.RUNNING); mockTask.handle(new TaskEventTAUpdate(mockTask.getLastAttempt().getID(), TaskEventType.T_ATTEMPT_SUCCEEDED)); assertTaskSucceededState(); eventHandler.events.clear(); mockTask.handle(new TaskEventTAUpdate(mockTask.getLastAttempt() .getID(), TaskEventType.T_ATTEMPT_FAILED)); assertTaskScheduledState(); Event event = eventHandler.events.get(0); Assert.assertEquals(AMNodeEventType.N_TA_ENDED, event.getType()); event = eventHandler.events.get(eventHandler.events.size()-1); Assert.assertEquals(VertexEventType.V_TASK_RESCHEDULED, event.getType()); }