@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 (InvalidStateTransitionException 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();
    }
  }
@Test
  /**
   * Kill map attempt for succeeded map task
   * {@link TaskState#SUCCEEDED}->{@link TaskState#SCHEDULED}
   */
  public void testKillAttemptForSuccessfulTask() {
    LOG.info("--- START: testKillAttemptForSuccessfulTask ---");
    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 TaskTAttemptKilledEvent(getLastAttempt().getAttemptId(), true));
    assertEquals(TaskAttemptEventType.TA_RESCHEDULE,
        taskAttemptEventHandler.lastTaskAttemptEvent.getType());
    assertTaskScheduledState();
  }