import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
public void verifyTaskGeneric(Task task, String id, String state,
    String type, String successfulAttempt, long startTime, long finishTime,
    long elapsedTime, float progress) {

  TaskId taskid = task.getID();
  String tid = MRApps.toString(taskid);
  TaskReport report = task.getReport();

  WebServicesTestUtils.checkStringMatch("id", tid, id);
  WebServicesTestUtils.checkStringMatch("type", task.getType().toString(),
      type);
  WebServicesTestUtils.checkStringMatch("state", report.getTaskState()
      .toString(), state);
  // not easily checked without duplicating logic, just make sure its here
  assertNotNull("successfulAttempt null", successfulAttempt);
  assertEquals("startTime wrong", report.getStartTime(), startTime);
  assertEquals("finishTime wrong", report.getFinishTime(), finishTime);
  assertEquals("elapsedTime wrong", finishTime - startTime, elapsedTime);
  assertEquals("progress wrong", report.getProgress() * 100, progress, 1e-3f);
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
private AtomicInteger containerNeed(TaskId taskID) {
  JobId jobID = taskID.getJobId();
  TaskType taskType = taskID.getTaskType();

  ConcurrentMap<JobId, AtomicInteger> relevantMap
      = taskType == TaskType.MAP ? mapContainerNeeds : reduceContainerNeeds;

  AtomicInteger result = relevantMap.get(jobID);

  if (result == null) {
    relevantMap.putIfAbsent(jobID, new AtomicInteger(0));
    result = relevantMap.get(jobID);
  }

  return result;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Test
//All Task attempts are timed out, leading to Job failure
public void testTimedOutTask() throws Exception {
  MRApp app = new TimeOutTaskMRApp(1, 0);
  Configuration conf = new Configuration();
  int maxAttempts = 2;
  conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, maxAttempts);
  // disable uberization (requires entire job to be reattempted, so max for
  // subtask attempts is overridden to 1)
  conf.setBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);
  Job job = app.submit(conf);
  app.waitForState(job, JobState.FAILED);
  Map<TaskId,Task> tasks = job.getTasks();
  Assert.assertEquals("Num tasks is not correct", 1, tasks.size());
  Task task = tasks.values().iterator().next();
  Assert.assertEquals("Task state not correct", TaskState.FAILED,
      task.getReport().getTaskState());
  Map<TaskAttemptId, TaskAttempt> attempts =
      tasks.values().iterator().next().getAttempts();
  Assert.assertEquals("Num attempts is not correct", maxAttempts,
      attempts.size());
  for (TaskAttempt attempt : attempts.values()) {
    Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
        attempt.getReport().getTaskAttemptState());
  }
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的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.api.records.TaskId; //导入依赖的package包/类
@Override
protected Collection<TaskAttempt> getTaskAttempts() {
  List<TaskAttempt> fewTaskAttemps = new ArrayList<TaskAttempt>();
  String taskTypeStr = $(TASK_TYPE);
  TaskType taskType = MRApps.taskType(taskTypeStr);
  String attemptStateStr = $(ATTEMPT_STATE);
  TaskAttemptStateUI neededState = MRApps
      .taskAttemptState(attemptStateStr);
  Job j = app.getJob();
  Map<TaskId, Task> tasks = j.getTasks(taskType);
  for (Task task : tasks.values()) {
    Map<TaskAttemptId, TaskAttempt> attempts = task.getAttempts();
    for (TaskAttempt attempt : attempts.values()) {
      if (neededState.correspondsTo(attempt.getState())) {
        fewTaskAttemps.add(attempt);
      }
    }
  }
  return fewTaskAttemps;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
public void verifyAMTask(JSONArray arr, Job job, String type)
    throws JSONException {
  for (Task task : job.getTasks().values()) {
    TaskId id = task.getID();
    String tid = MRApps.toString(id);
    Boolean found = false;
    if (type != null && task.getType() == MRApps.taskType(type)) {

      for (int i = 0; i < arr.length(); i++) {
        JSONObject info = arr.getJSONObject(i);
        if (tid.matches(info.getString("id"))) {
          found = true;
          verifyAMSingleTask(info, task);
        }
      }
      assertTrue("task with id: " + tid + " not in web service output", found);
    }
  }
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
CompletedTaskAttempt(TaskId taskId, TaskAttemptInfo attemptInfo) {
  this.attemptInfo = attemptInfo;
  this.attemptId = TypeConverter.toYarn(attemptInfo.getAttemptId());
  if (attemptInfo.getTaskStatus() != null) {
    this.state = TaskAttemptState.valueOf(attemptInfo.getTaskStatus());
  } else {
    this.state = TaskAttemptState.KILLED;
    localDiagMessage = "Attmpt state missing from History : marked as KILLED";
    diagnostics.add(localDiagMessage);
  }
  if (attemptInfo.getError() != null) {
    diagnostics.add(attemptInfo.getError());
  }
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
/**
 * Absorbs one TaskAttemptStatus
 *
 * @param reportedStatus the status report that we got from a task attempt
 *        that we want to fold into the speculation data for this job
 * @param timestamp the time this status corresponds to.  This matters
 *        because statuses contain progress.
 */
protected void statusUpdate(TaskAttemptStatus reportedStatus, long timestamp) {

  String stateString = reportedStatus.taskState.toString();

  TaskAttemptId attemptID = reportedStatus.id;
  TaskId taskID = attemptID.getTaskId();
  Job job = context.getJob(taskID.getJobId());

  if (job == null) {
    return;
  }

  Task task = job.getTask(taskID);

  if (task == null) {
    return;
  }

  estimator.updateAttempt(reportedStatus, timestamp);

  if (stateString.equals(TaskAttemptState.RUNNING.name())) {
    runningTasks.putIfAbsent(taskID, Boolean.TRUE);
  } else {
    runningTasks.remove(taskID, Boolean.TRUE);
    if (!stateString.equals(TaskAttemptState.STARTING.name())) {
      runningTaskAttemptStatistics.remove(attemptID);
    }
  }
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
/**
 * @return The end of the JS map that is the jquery datatable config for the
 * attempts table. 
 */
private String attemptsTableInit() {
  TaskType type = null;
  String symbol = $(TASK_TYPE);
  if (!symbol.isEmpty()) {
    type = MRApps.taskType(symbol);
  } else {
    TaskId taskID = MRApps.toTaskID($(TASK_ID));
    type = taskID.getTaskType();
  }
  StringBuilder b = tableInit()
    .append(", 'aaData': attemptsTableData")
    .append(", bDeferRender: true")
    .append(", bProcessing: true")
    .append("\n,aoColumnDefs:[\n")

    //logs column should not filterable (it includes container ID which may pollute searches)
    .append("\n{'aTargets': [ 4 ]")
    .append(", 'bSearchable': false }")

    .append("\n, {'sType':'numeric', 'aTargets': [ 0 ]")
    .append(", 'mRender': parseHadoopAttemptID }")

    .append("\n, {'sType':'numeric', 'aTargets': [ 5, 6")
    //Column numbers are different for maps and reduces
    .append(type == TaskType.REDUCE ? ", 7, 8" : "")
    .append(" ], 'mRender': renderHadoopDate }")

    .append("\n, {'sType':'numeric', 'aTargets': [")
    .append(type == TaskType.REDUCE ? "9, 10, 11, 12" : "7")
    .append(" ], 'mRender': renderHadoopElapsedTime }]")

    // Sort by id upon page load
    .append("\n, aaSorting: [[0, 'asc']]")
    .append("}");
    return b.toString();
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
boolean canSpeculate(AppContext context, TaskId taskID) {
  // This class rejects speculating any task that already has speculations,
  //  or isn't running.
  //  Subclasses should call TaskSpeculationPredicate.canSpeculate(...) , but
  //  can be even more restrictive.
  JobId jobID = taskID.getJobId();
  Job job = context.getJob(jobID);
  Task task = job.getTask(taskID);
  return task.getAttempts().size() == 1;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Override
public void setTaskId(TaskId taskId) {
  maybeInitBuilder();
  if (taskId == null) 
    builder.clearTaskId();
  this.taskId = taskId;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Override
public TaskId getTaskId() {
  GetTaskReportRequestProtoOrBuilder p = viaProto ? proto : builder;
  if (this.taskId != null) {
    return this.taskId;
  }
  if (!p.hasTaskId()) {
    return null;
  }
  this.taskId =  convertFromProtoFormat(p.getTaskId());
  return this.taskId;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Override
public long thresholdRuntime(TaskId taskID) {
  JobId jobID = taskID.getJobId();
  Job job = context.getJob(jobID);

  TaskType type = taskID.getTaskType();

  DataStatistics statistics
      = dataStatisticsForTask(taskID);

  int completedTasksOfType
      = type == TaskType.MAP
          ? job.getCompletedMaps() : job.getCompletedReduces();

  int totalTasksOfType
      = type == TaskType.MAP
          ? job.getTotalMaps() : job.getTotalReduces();

  if (completedTasksOfType < MINIMUM_COMPLETE_NUMBER_TO_SPECULATE
      || (((float)completedTasksOfType) / totalTasksOfType)
            < MINIMUM_COMPLETE_PROPORTION_TO_SPECULATE ) {
    return Long.MAX_VALUE;
  }

  long result =  statistics == null
      ? Long.MAX_VALUE
      : (long)statistics.outlier(slowTaskRelativeTresholds.get(job));
  return result;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Override
public long estimatedNewAttemptRuntime(TaskId id) {
  DataStatistics statistics = dataStatisticsForTask(id);

  if (statistics == null) {
    return -1L;
  }

  return (long)statistics.mean();
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
/**
 * Ensure that a TASK_ID was passed into the page.
 */
public void requireTask() {
  if ($(TASK_ID).isEmpty()) {
    badRequest("missing task ID");
    throw new RuntimeException("missing task ID");
  }

  TaskId taskID = MRApps.toTaskID($(TASK_ID));
  Job job = app.context.getJob(taskID.getJobId());
  app.setJob(job);
  if (app.getJob() == null) {
    notFound(MRApps.toString(taskID.getJobId()));
    throw new RuntimeException("Not Found: " + $(JOB_ID));
  } else {
    app.setTask(app.getJob().getTask(taskID));
    if (app.getTask() == null) {
      notFound($(TASK_ID));
      throw new RuntimeException("Not Found: " + $(TASK_ID));
    }
  }
  if (!checkAccess(job)) {
    accessDenied("User " + request().getRemoteUser() + " does not have " +
        " permission to view job " + $(JOB_ID));
    throw new RuntimeException("Access denied: User " +
        request().getRemoteUser() + " does not have permission to view job " +
        $(JOB_ID));
  }
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Override
public TaskId getTaskId() {
  TaskReportProtoOrBuilder p = viaProto ? proto : builder;
  if (this.taskId != null) {
    return this.taskId;
  }
  if (!p.hasTaskId()) {
    return null;
  }
  this.taskId = convertFromProtoFormat(p.getTaskId());
  return this.taskId;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
public TaskRecoverEvent(TaskId taskID, TaskInfo taskInfo,
    OutputCommitter committer, boolean recoverTaskOutput) {
  super(taskID, TaskEventType.T_RECOVER);
  this.taskInfo = taskInfo;
  this.committer = committer;
  this.recoverTaskOutput = recoverTaskOutput;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
private static JobCounterUpdateEvent createJobCounterUpdateEventTASucceeded(
    TaskAttemptImpl taskAttempt) {
  TaskId taskId = taskAttempt.attemptId.getTaskId();
  JobCounterUpdateEvent jce = new JobCounterUpdateEvent(taskId.getJobId());
  updateMillisCounters(jce, taskAttempt);
  return jce;
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Override
public Task getTask(TaskId taskID) {
  readLock.lock();
  try {
    return tasks.get(taskID);
  } finally {
    readLock.unlock();
  }
}
 

import org.apache.hadoop.mapreduce.v2.api.records.TaskId; //导入依赖的package包/类
@Test
public void testTaskFailWithUnusedContainer() throws Exception {
  MRApp app = new MRAppWithFailingTaskAndUnusedContainer();
  Configuration conf = new Configuration();
  int maxAttempts = 1;
  conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, maxAttempts);
  // disable uberization (requires entire job to be reattempted, so max for
  // subtask attempts is overridden to 1)
  conf.setBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);
  Job job = app.submit(conf);
  app.waitForState(job, JobState.RUNNING);
  Map<TaskId, Task> tasks = job.getTasks();
  Assert.assertEquals("Num tasks is not correct", 1, tasks.size());
  Task task = tasks.values().iterator().next();
  app.waitForState(task, TaskState.SCHEDULED);
  Map<TaskAttemptId, TaskAttempt> attempts = tasks.values().iterator()
      .next().getAttempts();
  Assert.assertEquals("Num attempts is not correct", maxAttempts, attempts
      .size());
  TaskAttempt attempt = attempts.values().iterator().next();
  app.waitForInternalState((TaskAttemptImpl) attempt,
      TaskAttemptStateInternal.ASSIGNED);
  app.getDispatcher().getEventHandler().handle(
      new TaskAttemptEvent(attempt.getID(),
          TaskAttemptEventType.TA_CONTAINER_COMPLETED));
  app.waitForState(job, JobState.FAILED);
}