TaskAttemptListenerImpl extends CompositeService implements TaskUmbilicalProtocol, TaskAttemptListener { @Override public AMFeedback statusUpdate(TaskAttemptID taskAttemptID, TaskStatus taskStatus) throws IOException, InterruptedException { org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId yarnAttemptID = TypeConverter.toYarn(taskAttemptID); AMFeedback feedback = new AMFeedback(); feedback.setTaskFound(true); AtomicReference<TaskAttemptStatus> lastStatusRef = attemptIdToStatus.get(yarnAttemptID); if (lastStatusRef == null) { if (!taskHeartbeatHandler.hasRecentlyUnregistered(yarnAttemptID)) { LOG.error("Status update was called with illegal TaskAttemptId: " + yarnAttemptID); feedback.setTaskFound(false); } return feedback; } if (getConfig().getBoolean(MRJobConfig.TASK_PREEMPTION, false) && preemptionPolicy.isPreempted(yarnAttemptID)) { feedback.setPreemption(true); LOG.info("Setting preemption bit for task: "+ yarnAttemptID + " of type " + yarnAttemptID.getTaskId().getTaskType()); } if (taskStatus == null) { if (LOG.isDebugEnabled()) { LOG.debug("Ping from " + taskAttemptID.toString()); } return feedback; } taskHeartbeatHandler.progressing(yarnAttemptID); TaskAttemptStatus taskAttemptStatus = new TaskAttemptStatus(); taskAttemptStatus.id = yarnAttemptID; taskAttemptStatus.progress = taskStatus.getProgress(); LOG.info("Progress of TaskAttempt " + taskAttemptID + " is : " + taskStatus.getProgress()); taskAttemptStatus.stateString = taskStatus.getStateString(); taskAttemptStatus.phase = TypeConverter.toYarn(taskStatus.getPhase()); taskAttemptStatus.counters = new org.apache.hadoop.mapreduce.Counters( taskStatus.getCounters()); if (taskStatus.getIsMap() && taskStatus.getMapFinishTime() != 0) { taskAttemptStatus.mapFinishTime = taskStatus.getMapFinishTime(); } if (!taskStatus.getIsMap() && taskStatus.getShuffleFinishTime() != 0) { taskAttemptStatus.shuffleFinishTime = taskStatus.getShuffleFinishTime(); } if (!taskStatus.getIsMap() && taskStatus.getSortFinishTime() != 0) { taskAttemptStatus.sortFinishTime = taskStatus.getSortFinishTime(); } if (taskStatus.getFetchFailedMaps() != null && taskStatus.getFetchFailedMaps().size() > 0) { taskAttemptStatus.fetchFailedMaps = new ArrayList<org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId>(); for (TaskAttemptID failedMapId : taskStatus.getFetchFailedMaps()) { taskAttemptStatus.fetchFailedMaps.add( TypeConverter.toYarn(failedMapId)); } } coalesceStatusUpdate(yarnAttemptID, taskAttemptStatus, lastStatusRef); return feedback; } TaskAttemptListenerImpl(AppContext context,
      JobTokenSecretManager jobTokenSecretManager,
      RMHeartbeatHandler rmHeartbeatHandler,
      AMPreemptionPolicy preemptionPolicy);  TaskAttemptListenerImpl(AppContext context,
      JobTokenSecretManager jobTokenSecretManager,
      RMHeartbeatHandler rmHeartbeatHandler,
      AMPreemptionPolicy preemptionPolicy, byte[] secretShuffleKey); @Override InetSocketAddress getAddress(); @Override boolean canCommit(TaskAttemptID taskAttemptID); @Override void commitPending(TaskAttemptID taskAttemptID, TaskStatus taskStatsu); @Override void preempted(TaskAttemptID taskAttemptID, TaskStatus taskStatus); @Override void done(TaskAttemptID taskAttemptID); @Override void fatalError(TaskAttemptID taskAttemptID, String msg, boolean fastFail); @Override void fsError(TaskAttemptID taskAttemptID, String message); @Override void shuffleError(TaskAttemptID taskAttemptID, String message); @Override MapTaskCompletionEventsUpdate getMapCompletionEvents(
      JobID jobIdentifier, int startIndex, int maxEvents,
      TaskAttemptID taskAttemptID); @Override void reportDiagnosticInfo(TaskAttemptID taskAttemptID, String diagnosticInfo); @Override AMFeedback statusUpdate(TaskAttemptID taskAttemptID,
      TaskStatus taskStatus); @Override long getProtocolVersion(String arg0, long arg1); @Override void reportNextRecordRange(TaskAttemptID taskAttemptID, Range range); @Override JvmTask getTask(JvmContext context); @Override void registerPendingTask(
      org.apache.hadoop.mapred.Task task, WrappedJvmID jvmID); @Override void registerLaunchedTask(
      org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID,
      WrappedJvmID jvmId); @Override void unregister(
      org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID,
      WrappedJvmID jvmID); @Override ProtocolSignature getProtocolSignature(String protocol,
      long clientVersion, int clientMethodsHash); @Override TaskCheckpointID getCheckpointID(TaskID taskId); @Override void setCheckpointID(TaskID taskId, TaskCheckpointID cid);  }
@Test public void testStatusUpdateProgress() throws IOException, InterruptedException { configureMocks(); startListener(true); verify(hbHandler).register(attemptId); AMFeedback feedback = listener.statusUpdate(attemptID, null); assertTrue(feedback.getTaskFound()); verify(hbHandler, never()).progressing(eq(attemptId)); MapTaskStatus mockStatus = new MapTaskStatus(attemptID, 0.0f, 1, TaskStatus.State.RUNNING, "", "RUNNING", "", TaskStatus.Phase.MAP, new Counters()); feedback = listener.statusUpdate(attemptID, mockStatus); assertTrue(feedback.getTaskFound()); verify(hbHandler).progressing(eq(attemptId)); }
@Test public void testSingleStatusUpdate() throws IOException, InterruptedException { configureMocks(); startListener(true); listener.statusUpdate(attemptID, firstReduceStatus); verify(ea).handle(eventCaptor.capture()); TaskAttemptStatusUpdateEvent updateEvent = (TaskAttemptStatusUpdateEvent) eventCaptor.getValue(); TaskAttemptStatus status = updateEvent.getTaskAttemptStatusRef().get(); assertTrue(status.fetchFailedMaps.contains(TASKATTEMPTID1)); assertEquals(1, status.fetchFailedMaps.size()); assertEquals(Phase.SHUFFLE, status.phase); }