@Test (timeout=10000) public void testGetMapCompletionEvents() throws IOException { TaskAttemptCompletionEvent[] empty = {}; TaskAttemptCompletionEvent[] taskEvents = { createTce(0, true, TaskAttemptCompletionEventStatus.OBSOLETE), createTce(1, false, TaskAttemptCompletionEventStatus.FAILED), createTce(2, true, TaskAttemptCompletionEventStatus.SUCCEEDED), createTce(3, false, TaskAttemptCompletionEventStatus.FAILED) }; TaskAttemptCompletionEvent[] mapEvents = { taskEvents[0], taskEvents[2] }; Job mockJob = mock(Job.class); when(mockJob.getTaskAttemptCompletionEvents(0, 100)) .thenReturn(taskEvents); when(mockJob.getTaskAttemptCompletionEvents(0, 2)) .thenReturn(Arrays.copyOfRange(taskEvents, 0, 2)); when(mockJob.getTaskAttemptCompletionEvents(2, 100)) .thenReturn(Arrays.copyOfRange(taskEvents, 2, 4)); when(mockJob.getMapAttemptCompletionEvents(0, 100)).thenReturn( TypeConverter.fromYarn(mapEvents)); when(mockJob.getMapAttemptCompletionEvents(0, 2)).thenReturn( TypeConverter.fromYarn(mapEvents)); when(mockJob.getMapAttemptCompletionEvents(2, 100)).thenReturn( TypeConverter.fromYarn(empty)); AppContext appCtx = mock(AppContext.class); when(appCtx.getJob(any(JobId.class))).thenReturn(mockJob); JobTokenSecretManager secret = mock(JobTokenSecretManager.class); RMHeartbeatHandler rmHeartbeatHandler = mock(RMHeartbeatHandler.class); final TaskHeartbeatHandler hbHandler = mock(TaskHeartbeatHandler.class); TaskAttemptListenerImpl listener = new MockTaskAttemptListenerImpl(appCtx, secret, rmHeartbeatHandler) { @Override protected void registerHeartbeatHandler(Configuration conf) { taskHeartbeatHandler = hbHandler; } }; Configuration conf = new Configuration(); listener.init(conf); listener.start(); JobID jid = new JobID("12345", 1); TaskAttemptID tid = new TaskAttemptID("12345", 1, TaskType.REDUCE, 1, 0); MapTaskCompletionEventsUpdate update = listener.getMapCompletionEvents(jid, 0, 100, tid); assertEquals(2, update.events.length); update = listener.getMapCompletionEvents(jid, 0, 2, tid); assertEquals(2, update.events.length); update = listener.getMapCompletionEvents(jid, 2, 100, tid); assertEquals(0, update.events.length); }
@Override public MapTaskCompletionEventsUpdate getMapCompletionEvents( JobID jobIdentifier, int startIndex, int maxEvents, TaskAttemptID taskAttemptID) throws IOException { LOG.info("MapCompletionEvents request from " + taskAttemptID.toString() + ". startIndex " + startIndex + " maxEvents " + maxEvents); boolean shouldReset = false; org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID = TypeConverter.toYarn(taskAttemptID); TaskCompletionEvent[] events = context.getJob(attemptID.getTaskId().getJobId()).getMapAttemptCompletionEvents( startIndex, maxEvents); taskHeartbeatHandler.progressing(attemptID); return new MapTaskCompletionEventsUpdate(events, shouldReset); }
TaskAttemptListenerImpl extends CompositeService implements TaskUmbilicalProtocol, TaskAttemptListener { @Override public MapTaskCompletionEventsUpdate getMapCompletionEvents( JobID jobIdentifier, int startIndex, int maxEvents, TaskAttemptID taskAttemptID) throws IOException { LOG.info("MapCompletionEvents request from " + taskAttemptID.toString() + ". startIndex " + startIndex + " maxEvents " + maxEvents); boolean shouldReset = false; org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID = TypeConverter.toYarn(taskAttemptID); TaskCompletionEvent[] events = context.getJob(attemptID.getTaskId().getJobId()).getMapAttemptCompletionEvents( startIndex, maxEvents); taskHeartbeatHandler.progressing(attemptID); return new MapTaskCompletionEventsUpdate(events, shouldReset); } }
TaskAttemptListenerImpl extends CompositeService implements TaskUmbilicalProtocol, TaskAttemptListener { @Override public MapTaskCompletionEventsUpdate getMapCompletionEvents( JobID jobIdentifier, int startIndex, int maxEvents, TaskAttemptID taskAttemptID) throws IOException { LOG.info("MapCompletionEvents request from " + taskAttemptID.toString() + ". startIndex " + startIndex + " maxEvents " + maxEvents); boolean shouldReset = false; org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID = TypeConverter.toYarn(taskAttemptID); TaskCompletionEvent[] events = context.getJob(attemptID.getTaskId().getJobId()).getMapAttemptCompletionEvents( startIndex, maxEvents); taskHeartbeatHandler.progressing(attemptID); return new MapTaskCompletionEventsUpdate(events, shouldReset); } TaskAttemptListenerImpl(AppContext context,
      JobTokenSecretManager jobTokenSecretManager,
      RMHeartbeatHandler rmHeartbeatHandler); }
TaskAttemptListenerImpl extends CompositeService implements TaskUmbilicalProtocol, TaskAttemptListener { @Override public MapTaskCompletionEventsUpdate getMapCompletionEvents( JobID jobIdentifier, int startIndex, int maxEvents, TaskAttemptID taskAttemptID) throws IOException { LOG.info("MapCompletionEvents request from " + taskAttemptID.toString() + ". startIndex " + startIndex + " maxEvents " + maxEvents); boolean shouldReset = false; org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID = TypeConverter.toYarn(taskAttemptID); TaskCompletionEvent[] events = context.getJob(attemptID.getTaskId().getJobId()).getMapAttemptCompletionEvents( startIndex, maxEvents); taskHeartbeatHandler.progressing(attemptID); return new MapTaskCompletionEventsUpdate(events, shouldReset); } TaskAttemptListenerImpl(AppContext context,
      JobTokenSecretManager jobTokenSecretManager,
      RMHeartbeatHandler rmHeartbeatHandler); @Override InetSocketAddress getAddress(); @Override boolean canCommit(TaskAttemptID taskAttemptID); @Override void commitPending(TaskAttemptID taskAttemptID, TaskStatus taskStatsu); @Override void done(TaskAttemptID taskAttemptID); @Override void fatalError(TaskAttemptID taskAttemptID, String msg); @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 boolean ping(TaskAttemptID taskAttemptID); @Override void reportDiagnosticInfo(TaskAttemptID taskAttemptID, String diagnosticInfo); @Override boolean 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); }
TaskAttemptListenerImpl extends CompositeService implements TaskUmbilicalProtocol, TaskAttemptListener { @Override public MapTaskCompletionEventsUpdate getMapCompletionEvents( JobID jobIdentifier, int startIndex, int maxEvents, TaskAttemptID taskAttemptID) throws IOException { LOG.info("MapCompletionEvents request from " + taskAttemptID.toString() + ". startIndex " + startIndex + " maxEvents " + maxEvents); boolean shouldReset = false; org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID = TypeConverter.toYarn(taskAttemptID); TaskCompletionEvent[] events = context.getJob(attemptID.getTaskId().getJobId()).getMapAttemptCompletionEvents( startIndex, maxEvents); taskHeartbeatHandler.progressing(attemptID); return new MapTaskCompletionEventsUpdate(events, shouldReset); } TaskAttemptListenerImpl(AppContext context,
      JobTokenSecretManager jobTokenSecretManager,
      RMHeartbeatHandler rmHeartbeatHandler); @Override InetSocketAddress getAddress(); @Override boolean canCommit(TaskAttemptID taskAttemptID); @Override void commitPending(TaskAttemptID taskAttemptID, TaskStatus taskStatsu); @Override void done(TaskAttemptID taskAttemptID); @Override void fatalError(TaskAttemptID taskAttemptID, String msg); @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 boolean ping(TaskAttemptID taskAttemptID); @Override void reportDiagnosticInfo(TaskAttemptID taskAttemptID, String diagnosticInfo); @Override boolean 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);  }