TaskHeartbeatHandler extends AbstractService { public void register(TaskAttemptId attemptID) { runningAttempts.put(attemptID, new ReportTime(clock.getTime())); } TaskHeartbeatHandler(EventHandler eventHandler, Clock clock,
      int numThreads); void progressing(TaskAttemptId attemptID); void register(TaskAttemptId attemptID); void unregister(TaskAttemptId attemptID); boolean hasRecentlyUnregistered(TaskAttemptId attemptID); @VisibleForTesting long getTaskTimeOut();  }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testTimeout() throws InterruptedException { EventHandler mockHandler = mock(EventHandler.class); Clock clock = SystemClock.getInstance(); TaskHeartbeatHandler hb = new TaskHeartbeatHandler(mockHandler, clock, 1); Configuration conf = new Configuration(); conf.setInt(MRJobConfig.TASK_TIMEOUT, 10); conf.setLong(MRJobConfig.TASK_PROGRESS_REPORT_INTERVAL, 5); conf.setInt(MRJobConfig.TASK_TIMEOUT_CHECK_INTERVAL_MS, 10); hb.init(conf); hb.start(); try { ApplicationId appId = ApplicationId.newInstance(0l, 5); JobId jobId = MRBuilderUtils.newJobId(appId, 4); TaskId tid = MRBuilderUtils.newTaskId(jobId, 3, TaskType.MAP); TaskAttemptId taid = MRBuilderUtils.newTaskAttemptId(tid, 2); hb.register(taid); Thread.sleep(100); verify(mockHandler, times(2)).handle(any(Event.class)); } finally { hb.stop(); } }