import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@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();
  }
}
 

import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@Override
public void transition(TaskImpl task, TaskEvent event) {
  TaskTAttemptEvent ev = (TaskTAttemptEvent) event;
  // The nextAttemptNumber is commit pending, decide on set the commitAttempt
  TaskAttemptId attemptID = ev.getTaskAttemptID();
  if (task.commitAttempt == null) {
    // TODO: validate attemptID
    task.commitAttempt = attemptID;
    LOG.info(attemptID + " given a go for committing the task output.");
  } else {
    // Don't think this can be a pluggable decision, so simply raise an
    // event for the TaskAttempt to delete its output.
    LOG.info(task.commitAttempt
        + " already given a go for committing the task output, so killing "
        + attemptID);
    task.eventHandler.handle(new TaskAttemptKillEvent(attemptID,
        SPECULATION + task.commitAttempt + " committed first!"));
  }
}
 

import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public KillTaskResponse killTask(KillTaskRequest request) 
  throws IOException {
  TaskId taskId = request.getTaskId();
  UserGroupInformation callerUGI = UserGroupInformation.getCurrentUser();
  String message = "Kill task " + taskId + " received from " + callerUGI
      + " at " + Server.getRemoteAddress();
  LOG.info(message);
  verifyAndGetTask(taskId, JobACL.MODIFY_JOB);
  appContext.getEventHandler().handle(
      new TaskEvent(taskId, TaskEventType.T_KILL));
  KillTaskResponse response = 
    recordFactory.newRecordInstance(KillTaskResponse.class);
  return response;
}
 

import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@Test
public void testJobError() throws Exception {
  MRApp app = new MRApp(1, 0, false, this.getClass().getName(), true);
  Job job = app.submit(new Configuration());
  app.waitForState(job, JobState.RUNNING);
  Assert.assertEquals("Num tasks not correct", 1, job.getTasks().size());
  Iterator<Task> it = job.getTasks().values().iterator();
  Task task = it.next();
  app.waitForState(task, TaskState.RUNNING);

  //send an invalid event on task at current state
  app.getContext().getEventHandler().handle(
      new TaskEvent(
          task.getID(), TaskEventType.T_SCHEDULE));

  //this must lead to job error
  app.waitForState(job, JobState.ERROR);
}
 

import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@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();
  }
}
 

import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@Override
public void transition(TaskImpl task, TaskEvent event) {
  TaskAttemptId taskAttemptId =
      ((TaskTAttemptEvent) event).getTaskAttemptID();
  task.handleTaskAttemptCompletion(
      taskAttemptId, 
      TaskAttemptCompletionEventStatus.KILLED);
  task.finishedAttempts.add(taskAttemptId);
  task.inProgressAttempts.remove(taskAttemptId);
  if (task.successfulAttempt == null) {
    boolean rescheduleNewAttempt = false;
    if (event instanceof TaskTAttemptKilledEvent) {
      rescheduleNewAttempt =
          ((TaskTAttemptKilledEvent)event).getRescheduleAttempt();
    }
    task.addAndScheduleAttempt(Avataar.VIRGIN, rescheduleNewAttempt);
  }
  if ((task.commitAttempt != null) && (task.commitAttempt == taskAttemptId)) {
	task.commitAttempt = null;
  }
}
 

import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent; //导入依赖的package包/类
@Override
public void transition(TaskImpl task, TaskEvent event) {
  TaskTAttemptEvent taskTAttemptEvent = (TaskTAttemptEvent) event;
  TaskAttemptId taskAttemptId = taskTAttemptEvent.getTaskAttemptID();
  task.handleTaskAttemptCompletion(
      taskAttemptId, 
      TaskAttemptCompletionEventStatus.SUCCEEDED);
  task.finishedAttempts.add(taskAttemptId);
  task.inProgressAttempts.remove(taskAttemptId);
  task.successfulAttempt = taskAttemptId;
  task.sendTaskSucceededEvents();
  for (TaskAttempt attempt : task.attempts.values()) {
    if (attempt.getID() != task.successfulAttempt &&
        // This is okay because it can only talk us out of sending a
        //  TA_KILL message to an attempt that doesn't need one for
        //  other reasons.
        !attempt.isFinished()) {
      LOG.info("Issuing kill to other attempt " + attempt.getID());
      task.eventHandler.handle(
          new TaskAttemptEvent(attempt.getID(), 
              TaskAttemptEventType.TA_KILL));
    }
  }
  task.finished(TaskStateInternal.SUCCEEDED);
}