Java 类org.apache.hadoop.mapreduce.jobhistory.TaskFinishedEvent 实例源码

项目：FlexMap 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
JobTaskEvent jobTaskEvent = new JobTaskEvent(taskId, TaskState.SUCCEEDED);
long    totalTime   = this.getFinishTime() - this.getLaunchTime();
long    HDFSRecords = this.getSuccessfulAttempt().getCounters().findCounter(TaskCounter.MAP_INPUT_RECORDS).getValue();
long    executionTime  = this.getSuccessfulAttempt().getEndExecutionTime() - this.getSuccessfulAttempt().getBeginExecutionTime();
double  executionSpeed = HDFSRecords*1.0 / executionTime*1.0;
double  executionRatio = 1.0*executionTime/ totalTime;
LOG.info("inform");
LOG.info("hdfsRecrds:"+HDFSRecords);
LOG.info("excutuinTime:"+executionTime);
LOG.info("totalTime:"+executionTime);
LOG.info("excutionSpeed:"+executionSpeed);
LOG.info("excutionRatio:"+executionRatio);
LOG.info("host:"+this.getSuccessfulAttempt().getNodeId().getHost());
LOG.info("/inform");
jobTaskEvent.setTaskExecutionTime((long)executionSpeed);
jobTaskEvent.setTaskExecutionRatio(executionRatio);
jobTaskEvent.setAttemptId(successfulAttempt);
   eventHandler.handle(jobTaskEvent);
   if (historyTaskStartGenerated) {
     TaskFinishedEvent tfe = createTaskFinishedEvent(this,
         TaskStateInternal.SUCCEEDED);
     eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
   }
 }

项目：hadoop 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hadoop 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hadoop 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hadoop 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：aliyun-oss-hadoop-fs 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：aliyun-oss-hadoop-fs 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：aliyun-oss-hadoop-fs 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：aliyun-oss-hadoop-fs 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：big-c 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：big-c 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：big-c 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：big-c 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hadoop-2.6.0-cdh5.4.3 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hadoop-2.6.0-cdh5.4.3 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：hadoop-plus 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hadoop-plus 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hadoop-plus 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hadoop-plus 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：FlexMap 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hops 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hops 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hops 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hops 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：hadoop-TCP 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hadoop-TCP 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hadoop-TCP 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hadoop-TCP 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：hardfs 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hardfs 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hardfs 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hardfs 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：hadoop-on-lustre2 文件：TaskImpl.java

private static TaskFinishedEvent createTaskFinishedEvent(TaskImpl task, TaskStateInternal taskState) {
  TaskFinishedEvent tfe =
    new TaskFinishedEvent(TypeConverter.fromYarn(task.taskId),
      TypeConverter.fromYarn(task.successfulAttempt),
      task.getFinishTime(task.successfulAttempt),
      TypeConverter.fromYarn(task.taskId.getTaskType()),
      taskState.toString(),
      task.getCounters());
  return tfe;
}

项目：hadoop-on-lustre2 文件：TaskImpl.java

private void sendTaskSucceededEvents() {
  eventHandler.handle(new JobTaskEvent(taskId, TaskState.SUCCEEDED));
  LOG.info("Task succeeded with attempt " + successfulAttempt);
  if (historyTaskStartGenerated) {
    TaskFinishedEvent tfe = createTaskFinishedEvent(this,
        TaskStateInternal.SUCCEEDED);
    eventHandler.handle(new JobHistoryEvent(taskId.getJobId(), tfe));
  }
}

项目：hadoop-on-lustre2 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, null, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：hadoop-on-lustre2 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  ParsedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}

项目：mapreduce-fork 文件：Task20LineHistoryEventEmitter.java

HistoryEvent maybeEmitEvent(ParsedLine line, String taskIDName,
    HistoryEventEmitter thatg) {
  if (taskIDName == null) {
    return null;
  }

  TaskID taskID = TaskID.forName(taskIDName);

  String status = line.get("TASK_STATUS");
  String finishTime = line.get("FINISH_TIME");

  String error = line.get("ERROR");

  String counters = line.get("COUNTERS");

  if (finishTime != null && error == null
      && (status != null && status.equalsIgnoreCase("success"))) {
    Counters eventCounters = maybeParseCounters(counters);

    Task20LineHistoryEventEmitter that =
        (Task20LineHistoryEventEmitter) thatg;

    if (that.originalTaskType == null) {
      return null;
    }

    return new TaskFinishedEvent(taskID, Long.parseLong(finishTime),
        that.originalTaskType, status, eventCounters);
  }

  return null;
}

项目：mapreduce-fork 文件：JobBuilder.java

private void processTaskFinishedEvent(TaskFinishedEvent event) {
  LoggedTask task =
      getOrMakeTask(event.getTaskType(), event.getTaskId().toString(), false);
  if (task == null) {
    return;
  }
  task.setFinishTime(event.getFinishTime());
  task.setTaskStatus(getPre21Value(event.getTaskStatus()));
  task.incorporateCounters(((TaskFinished) event.getDatum()).counters);
}