public void verifyAMTaskAttemptsXML(NodeList nodes, Task task) {
  assertEquals("incorrect number of elements", 1, nodes.getLength());

  for (TaskAttempt att : task.getAttempts().values()) {
    TaskAttemptId id = att.getID();
    String attid = MRApps.toString(id);
    Boolean found = false;
    for (int i = 0; i < nodes.getLength(); i++) {
      Element element = (Element) nodes.item(i);

      if (attid.matches(WebServicesTestUtils.getXmlString(element, "id"))) {
        found = true;
        verifyAMTaskAttemptXML(element, att, task.getType());
      }
    }
    assertTrue("task with id: " + attid + " not in web service output", found);
  }
}
 

private ContainerRequestEvent
    createReq(JobId jobId, int taskAttemptId, int memory, String[] hosts,
        boolean earlierFailedAttempt, boolean reduce) {
  TaskId taskId;
  if (reduce) {
    taskId = MRBuilderUtils.newTaskId(jobId, 0, TaskType.REDUCE);
  } else {
    taskId = MRBuilderUtils.newTaskId(jobId, 0, TaskType.MAP);
  }
  TaskAttemptId attemptId = MRBuilderUtils.newTaskAttemptId(taskId,
      taskAttemptId);
  Resource containerNeed = Resource.newInstance(memory, 1);
  if (earlierFailedAttempt) {
    return ContainerRequestEvent
        .createContainerRequestEventForFailedContainer(attemptId,
            containerNeed);
  }
  return new ContainerRequestEvent(attemptId, containerNeed, hosts,
      new String[] { NetworkTopology.DEFAULT_RACK });
}
 

private void testMRAppHistory(MRApp app) throws Exception {
  Configuration conf = new Configuration();
  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", 4, attempts.size());

  Iterator<TaskAttempt> it = attempts.values().iterator();
  TaskAttemptReport report = it.next().getReport();
  Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
      report.getTaskAttemptState());
  Assert.assertEquals("Diagnostic Information is not Correct",
      "Test Diagnostic Event", report.getDiagnosticInfo());
  report = it.next().getReport();
  Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
      report.getTaskAttemptState());
}
 

private void updateStatus(MRApp app, TaskAttempt attempt, Phase phase) {
  TaskAttemptStatusUpdateEvent.TaskAttemptStatus status = new TaskAttemptStatusUpdateEvent.TaskAttemptStatus();
  status.counters = new Counters();
  status.fetchFailedMaps = new ArrayList<TaskAttemptId>();
  status.id = attempt.getID();
  status.mapFinishTime = 0;
  status.phase = phase;
  status.progress = 0.5f;
  status.shuffleFinishTime = 0;
  status.sortFinishTime = 0;
  status.stateString = "OK";
  status.taskState = attempt.getState();
  TaskAttemptStatusUpdateEvent event = new TaskAttemptStatusUpdateEvent(attempt.getID(),
      status);
  app.getContext().getEventHandler().handle(event);
}
 

private void testMRAppHistory(MRApp app) throws Exception {
  Configuration conf = new Configuration();
  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", 4, attempts.size());

  Iterator<TaskAttempt> it = attempts.values().iterator();
  TaskAttemptReport report = it.next().getReport();
  Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
      report.getTaskAttemptState());
  Assert.assertEquals("Diagnostic Information is not Correct",
      "Test Diagnostic Event", report.getDiagnosticInfo());
  report = it.next().getReport();
  Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
      report.getTaskAttemptState());
}
 

public void verifyAMTaskAttemptsXML(NodeList nodes, Task task) {
  assertEquals("incorrect number of elements", 1, nodes.getLength());

  for (TaskAttempt att : task.getAttempts().values()) {
    TaskAttemptId id = att.getID();
    String attid = MRApps.toString(id);
    Boolean found = false;
    for (int i = 0; i < nodes.getLength(); i++) {
      Element element = (Element) nodes.item(i);

      if (attid.matches(WebServicesTestUtils.getXmlString(element, "id"))) {
        found = true;
        verifyAMTaskAttemptXML(element, att, task.getType());
      }
    }
    assertTrue("task with id: " + attid + " not in web service output", found);
  }
}
 

private long storedPerAttemptValue
     (Map<TaskAttempt, AtomicLong> data, TaskAttemptId attemptID) {
  TaskId taskID = attemptID.getTaskId();
  JobId jobID = taskID.getJobId();
  Job job = context.getJob(jobID);

  Task task = job.getTask(taskID);

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

  TaskAttempt taskAttempt = task.getAttempt(attemptID);

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

  AtomicLong estimate = data.get(taskAttempt);

  return estimate == null ? -1L : estimate.get();

}
 

@Override
public GetTaskAttemptReportResponse getTaskAttemptReport(
    GetTaskAttemptReportRequest request) throws IOException {
  TaskAttemptId taskAttemptId = request.getTaskAttemptId();
  Job job = verifyAndGetJob(taskAttemptId.getTaskId().getJobId(), true);
  GetTaskAttemptReportResponse response = recordFactory.newRecordInstance(GetTaskAttemptReportResponse.class);
  response.setTaskAttemptReport(job.getTask(taskAttemptId.getTaskId()).getAttempt(taskAttemptId).getReport());
  return response;
}
 

private ContainerAllocatorEvent createDeallocateEvent(JobId jobId,
    int taskAttemptId, boolean reduce) {
  TaskId taskId;
  if (reduce) {
    taskId = MRBuilderUtils.newTaskId(jobId, 0, TaskType.REDUCE);
  } else {
    taskId = MRBuilderUtils.newTaskId(jobId, 0, TaskType.MAP);
  }
  TaskAttemptId attemptId =
      MRBuilderUtils.newTaskAttemptId(taskId, taskAttemptId);
  return new ContainerAllocatorEvent(attemptId,
      ContainerAllocator.EventType.CONTAINER_DEALLOCATE);
}
 

@Override
public void addAllRunningAttempts(final List<TaskAttemptId> runningAttempts) {
  if (runningAttempts == null)
    return;
  initRunningAttempts();
  this.runningAttempts.addAll(runningAttempts);
}
 

@Test
public void testTaskAttemptIdXMLCounters() throws JSONException, Exception {
  WebResource r = resource();
  Map<JobId, Job> jobsMap = appContext.getAllJobs();
  for (JobId id : jobsMap.keySet()) {
    String jobId = MRApps.toString(id);
    for (Task task : jobsMap.get(id).getTasks().values()) {

      String tid = MRApps.toString(task.getID());
      for (TaskAttempt att : task.getAttempts().values()) {
        TaskAttemptId attemptid = att.getID();
        String attid = MRApps.toString(attemptid);

        ClientResponse response = r.path("ws").path("v1").path("history")
            .path("mapreduce").path("jobs").path(jobId).path("tasks")
            .path(tid).path("attempts").path(attid).path("counters")
            .accept(MediaType.APPLICATION_XML).get(ClientResponse.class);

        assertEquals(MediaType.APPLICATION_XML_TYPE, response.getType());
        String xml = response.getEntity(String.class);
        DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
        DocumentBuilder db = dbf.newDocumentBuilder();
        InputSource is = new InputSource();
        is.setCharacterStream(new StringReader(xml));
        Document dom = db.parse(is);
        NodeList nodes = dom.getElementsByTagName("jobTaskAttemptCounters");

        verifyHsTaskCountersXML(nodes, att);
      }
    }
  }
}
 

@Override
public GetTaskReportResponse getTaskReport(GetTaskReportRequest request)
    throws IOException {
  GetTaskReportResponse resp =
    recordFactory.newRecordInstance(GetTaskReportResponse.class);
  TaskReport report = recordFactory.newRecordInstance(TaskReport.class);
  report.setTaskId(request.getTaskId());
  report.setTaskState(TaskState.NEW);
  Counters counters = recordFactory.newRecordInstance(Counters.class);
  counters.addAllCounterGroups(new HashMap<String, CounterGroup>());
  report.setCounters(counters);
  report.addAllRunningAttempts(new ArrayList<TaskAttemptId>());
  return resp;
}
 

@Override
public TaskAttemptId getAttemptId() {
  TaskAttemptCompletionEventProtoOrBuilder p = viaProto ? proto : builder;
  if (this.taskAttemptId != null) {
    return this.taskAttemptId;
  }
  if (!p.hasAttemptId()) {
    return null;
  }
  this.taskAttemptId = convertFromProtoFormat(p.getAttemptId());
  return this.taskAttemptId;
}
 

private void addRunningAttemptsToProto() {
  maybeInitBuilder();
  builder.clearRunningAttempts();
  if (runningAttempts == null)
    return;
  Iterable<TaskAttemptIdProto> iterable = new Iterable<TaskAttemptIdProto>() {
    @Override
    public Iterator<TaskAttemptIdProto> iterator() {
      return new Iterator<TaskAttemptIdProto>() {

        Iterator<TaskAttemptId> iter = runningAttempts.iterator();

        @Override
        public boolean hasNext() {
          return iter.hasNext();
        }

        @Override
        public TaskAttemptIdProto next() {
          return convertToProtoFormat(iter.next());
        }

        @Override
        public void remove() {
          throw new UnsupportedOperationException();

        }
      };

    }
  };
  builder.addAllRunningAttempts(iterable);
}
 

@Test
public void testTaskAttemptIdXMLCounters() throws JSONException, Exception {
  WebResource r = resource();
  Map<JobId, Job> jobsMap = appContext.getAllJobs();
  for (JobId id : jobsMap.keySet()) {
    String jobId = MRApps.toString(id);
    for (Task task : jobsMap.get(id).getTasks().values()) {

      String tid = MRApps.toString(task.getID());
      for (TaskAttempt att : task.getAttempts().values()) {
        TaskAttemptId attemptid = att.getID();
        String attid = MRApps.toString(attemptid);

        ClientResponse response = r.path("ws").path("v1").path("history")
            .path("mapreduce").path("jobs").path(jobId).path("tasks")
            .path(tid).path("attempts").path(attid).path("counters")
            .accept(MediaType.APPLICATION_XML).get(ClientResponse.class);

        assertEquals(MediaType.APPLICATION_XML_TYPE, response.getType());
        String xml = response.getEntity(String.class);
        DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
        DocumentBuilder db = dbf.newDocumentBuilder();
        InputSource is = new InputSource();
        is.setCharacterStream(new StringReader(xml));
        Document dom = db.parse(is);
        NodeList nodes = dom.getElementsByTagName("jobTaskAttemptCounters");

        verifyHsTaskCountersXML(nodes, att);
      }
    }
  }
}
 

@Override
public void setAttemptId(TaskAttemptId attemptId) {
  maybeInitBuilder();
  if (attemptId == null) 
    builder.clearAttemptId();
  this.taskAttemptId = attemptId;
}
 

ContainerId get(TaskAttemptId tId) {
  Container taskContainer;
  if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
    taskContainer = maps.get(tId);
  } else {
    taskContainer = reduces.get(tId);
  }

  if (taskContainer == null) {
    return null;
  } else {
    return taskContainer.getId();
  }
}
 

public void verifyTaskAttemptGeneric(TaskAttempt ta, TaskType ttype,
    String id, String state, String type, String rack,
    String nodeHttpAddress, String diagnostics, String assignedContainerId,
    long startTime, long finishTime, long elapsedTime, float progress) {

  TaskAttemptId attid = ta.getID();
  String attemptId = MRApps.toString(attid);

  WebServicesTestUtils.checkStringMatch("id", attemptId, id);
  WebServicesTestUtils.checkStringMatch("type", ttype.toString(), type);
  WebServicesTestUtils.checkStringMatch("state", ta.getState().toString(),
      state);
  WebServicesTestUtils.checkStringMatch("rack", ta.getNodeRackName(), rack);
  WebServicesTestUtils.checkStringMatch("nodeHttpAddress",
      ta.getNodeHttpAddress(), nodeHttpAddress);

  String expectDiag = "";
  List<String> diagnosticsList = ta.getDiagnostics();
  if (diagnosticsList != null && !diagnostics.isEmpty()) {
    StringBuffer b = new StringBuffer();
    for (String diag : diagnosticsList) {
      b.append(diag);
    }
    expectDiag = b.toString();
  }
  WebServicesTestUtils.checkStringMatch("diagnostics", expectDiag,
      diagnostics);
  WebServicesTestUtils.checkStringMatch("assignedContainerId",
      ConverterUtils.toString(ta.getAssignedContainerID()),
      assignedContainerId);

  assertEquals("startTime wrong", ta.getLaunchTime(), startTime);
  assertEquals("finishTime wrong", ta.getFinishTime(), finishTime);
  assertEquals("elapsedTime wrong", finishTime - startTime, elapsedTime);
  assertEquals("progress wrong", ta.getProgress() * 100, progress, 1e-3f);
}
 

public TaskAttemptRecoverEvent(TaskAttemptId id, TaskAttemptInfo taInfo,
    OutputCommitter committer, boolean recoverOutput) {
  super(id, TaskAttemptEventType.TA_RECOVER);
  this.taInfo = taInfo;
  this.committer = committer;
  this.recoverAttemptOutput = recoverOutput;
}
 

/**
 * Within the _local_ filesystem (not HDFS), all activity takes place within
 * a subdir inside one of the LOCAL_DIRS
 * (${local.dir}/usercache/$user/appcache/$appId/$contId/),
 * and all sub-MapTasks create the same filename ("file.out").  Rename that
 * to something unique (e.g., "map_0.out") to avoid possible collisions.
 *
 * Longer-term, we'll modify [something] to use TaskAttemptID-based
 * filenames instead of "file.out". (All of this is entirely internal,
 * so there are no particular compatibility issues.)
 */
@VisibleForTesting
protected static MapOutputFile renameMapOutputForReduce(JobConf conf,
    TaskAttemptId mapId, MapOutputFile subMapOutputFile) throws IOException {
  FileSystem localFs = FileSystem.getLocal(conf);
  // move map output to reduce input
  Path mapOut = subMapOutputFile.getOutputFile();
  FileStatus mStatus = localFs.getFileStatus(mapOut);
  Path reduceIn = subMapOutputFile.getInputFileForWrite(
      TypeConverter.fromYarn(mapId).getTaskID(), mStatus.getLen());
  Path mapOutIndex = subMapOutputFile.getOutputIndexFile();
  Path reduceInIndex = new Path(reduceIn.toString() + ".index");
  if (LOG.isDebugEnabled()) {
    LOG.debug("Renaming map output file for task attempt "
        + mapId.toString() + " from original location " + mapOut.toString()
        + " to destination " + reduceIn.toString());
  }
  if (!localFs.mkdirs(reduceIn.getParent())) {
    throw new IOException("Mkdirs failed to create "
        + reduceIn.getParent().toString());
  }
  if (!localFs.rename(mapOut, reduceIn))
    throw new IOException("Couldn't rename " + mapOut);
  if (!localFs.rename(mapOutIndex, reduceInIndex))
    throw new IOException("Couldn't rename " + mapOutIndex);

  return new RenamedMapOutputFile(reduceIn);
}
 

@Test
public void testTaskAttemptIdDefault() throws JSONException, Exception {
  WebResource r = resource();
  Map<JobId, Job> jobsMap = appContext.getAllJobs();

  for (JobId id : jobsMap.keySet()) {
    String jobId = MRApps.toString(id);

    for (Task task : jobsMap.get(id).getTasks().values()) {
      String tid = MRApps.toString(task.getID());

      for (TaskAttempt att : task.getAttempts().values()) {
        TaskAttemptId attemptid = att.getID();
        String attid = MRApps.toString(attemptid);

        ClientResponse response = r.path("ws").path("v1").path("history")
            .path("mapreduce").path("jobs").path(jobId).path("tasks")
            .path(tid).path("attempts").path(attid).get(ClientResponse.class);
        assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
        JSONObject json = response.getEntity(JSONObject.class);
        assertEquals("incorrect number of elements", 1, json.length());
        JSONObject info = json.getJSONObject("taskAttempt");
        verifyHsTaskAttempt(info, att, task.getType());
      }
    }
  }
}
 

@Test
public void testTaskAttemptIdXMLCounters() throws JSONException, Exception {
  WebResource r = resource();
  Map<JobId, Job> jobsMap = appContext.getAllJobs();
  for (JobId id : jobsMap.keySet()) {
    String jobId = MRApps.toString(id);
    for (Task task : jobsMap.get(id).getTasks().values()) {

      String tid = MRApps.toString(task.getID());
      for (TaskAttempt att : task.getAttempts().values()) {
        TaskAttemptId attemptid = att.getID();
        String attid = MRApps.toString(attemptid);

        ClientResponse response = r.path("ws").path("v1").path("mapreduce")
            .path("jobs").path(jobId).path("tasks").path(tid)
            .path("attempts").path(attid).path("counters")
            .accept(MediaType.APPLICATION_XML).get(ClientResponse.class);

        assertEquals(MediaType.APPLICATION_XML_TYPE, response.getType());
        String xml = response.getEntity(String.class);
        DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
        DocumentBuilder db = dbf.newDocumentBuilder();
        InputSource is = new InputSource();
        is.setCharacterStream(new StringReader(xml));
        Document dom = db.parse(is);
        NodeList nodes = dom.getElementsByTagName("jobTaskAttemptCounters");

        verifyAMTaskCountersXML(nodes, att);
      }
    }
  }
}
 

private ContainerFailedEvent createFailEvent(JobId jobId, int taskAttemptId,
    String host, boolean reduce) {
  TaskId taskId;
  if (reduce) {
    taskId = MRBuilderUtils.newTaskId(jobId, 0, TaskType.REDUCE);
  } else {
    taskId = MRBuilderUtils.newTaskId(jobId, 0, TaskType.MAP);
  }
  TaskAttemptId attemptId = MRBuilderUtils.newTaskAttemptId(taskId,
      taskAttemptId);
  return new ContainerFailedEvent(attemptId, host);    
}
 

@Override
public GetTaskReportResponse getTaskReport(GetTaskReportRequest request)
    throws IOException {
  GetTaskReportResponse resp =
    recordFactory.newRecordInstance(GetTaskReportResponse.class);
  TaskReport report = recordFactory.newRecordInstance(TaskReport.class);
  report.setTaskId(request.getTaskId());
  report.setTaskState(TaskState.NEW);
  Counters counters = recordFactory.newRecordInstance(Counters.class);
  counters.addAllCounterGroups(new HashMap<String, CounterGroup>());
  report.setCounters(counters);
  report.addAllRunningAttempts(new ArrayList<TaskAttemptId>());
  return resp;
}
 

public static TaskAttemptId toYarn(
    org.apache.hadoop.mapred.TaskAttemptID id) {
  TaskAttemptId taskAttemptId = recordFactory.newRecordInstance(TaskAttemptId.class);
  taskAttemptId.setTaskId(toYarn(id.getTaskID()));
  taskAttemptId.setId(id.getId());
  return taskAttemptId;
}
 

protected void attemptLaunched(TaskAttemptId attemptID) {
  if (autoComplete) {
    // send the done event
    getContext().getEventHandler().handle(
        new TaskAttemptEvent(attemptID,
            TaskAttemptEventType.TA_DONE));
  }
}
 

boolean remove(TaskAttemptId tId) {
  ContainerRequest req = null;
  if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
    req = maps.remove(tId);
  } else {
    req = reduces.remove(tId);
  }
  
  if (req == null) {
    return false;
  } else {
    decContainerReq(req);
    return true;
  }
}
 

@Override
public TaskAttemptId getSuccessfulAttempt() {
  TaskReportProtoOrBuilder p = viaProto ? proto : builder;
  if (this.successfulAttemptId != null) {
    return this.successfulAttemptId;
  }
  if (!p.hasSuccessfulAttempt()) {
    return null;
  }
  this.successfulAttemptId = convertFromProtoFormat(p.getSuccessfulAttempt());
  return this.successfulAttemptId;
}
 

@Test
public void testTaskAttemptId() throws JSONException, Exception {
  WebResource r = resource();
  Map<JobId, Job> jobsMap = appContext.getAllJobs();

  for (JobId id : jobsMap.keySet()) {
    String jobId = MRApps.toString(id);

    for (Task task : jobsMap.get(id).getTasks().values()) {
      String tid = MRApps.toString(task.getID());

      for (TaskAttempt att : task.getAttempts().values()) {
        TaskAttemptId attemptid = att.getID();
        String attid = MRApps.toString(attemptid);

        ClientResponse response = r.path("ws").path("v1").path("mapreduce")
            .path("jobs").path(jobId).path("tasks").path(tid)
            .path("attempts").path(attid).accept(MediaType.APPLICATION_JSON)
            .get(ClientResponse.class);
        assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
        JSONObject json = response.getEntity(JSONObject.class);
        assertEquals("incorrect number of elements", 1, json.length());
        JSONObject info = json.getJSONObject("taskAttempt");
        verifyAMTaskAttempt(info, att, task.getType());
      }
    }
  }
}
 

@Test
public void testTaskAttemptIdCounters() throws JSONException, Exception {
  WebResource r = resource();
  Map<JobId, Job> jobsMap = appContext.getAllJobs();

  for (JobId id : jobsMap.keySet()) {
    String jobId = MRApps.toString(id);

    for (Task task : jobsMap.get(id).getTasks().values()) {
      String tid = MRApps.toString(task.getID());

      for (TaskAttempt att : task.getAttempts().values()) {
        TaskAttemptId attemptid = att.getID();
        String attid = MRApps.toString(attemptid);

        ClientResponse response = r.path("ws").path("v1").path("history")
            .path("mapreduce").path("jobs").path(jobId).path("tasks")
            .path(tid).path("attempts").path(attid).path("counters")
            .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
        assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
        JSONObject json = response.getEntity(JSONObject.class);
        assertEquals("incorrect number of elements", 1, json.length());
        JSONObject info = json.getJSONObject("jobTaskAttemptCounters");
        verifyHsJobTaskAttemptCounters(info, att);
      }
    }
  }
}