### Question:
JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job, 
  EventHandler<JobEvent> { static JobState checkJobCompleteSuccess(JobImpl job) { if (job.completedTaskCount == job.getTasks().size()) { try { job.getCommitter().commitJob(job.getJobContext()); } catch (IOException e) { LOG.warn("Could not do commit for Job", e); } job.logJobHistoryFinishedEvent(); return job.finished(JobState.SUCCEEDED); } return null; } JobImpl(ApplicationId appID, Configuration conf,
      EventHandler eventHandler, TaskAttemptListener taskAttemptListener,
      JobTokenSecretManager jobTokenSecretManager,
      Credentials fsTokenCredentials, Clock clock, int startCount, 
      Set<TaskId> completedTasksFromPreviousRun, MRAppMetrics metrics,
      String userName); @Override JobId getID(); @Override boolean checkAccess(UserGroupInformation callerUGI, 
      JobACL jobOperation); @Override Task getTask(TaskId taskID); @Override int getCompletedMaps(); @Override int getCompletedReduces(); @Override boolean isUber(); @Override Counters getCounters(); static Counters newCounters(); static Counters incrTaskCounters(Counters counters,
                                          Collection<Task> tasks); static void incrAllCounters(Counters counters, Counters other); @Override TaskAttemptCompletionEvent[] getTaskAttemptCompletionEvents(
      int fromEventId, int maxEvents); @Override List<String> getDiagnostics(); @Override JobReport getReport(); @Override Map<TaskId, Task> getTasks(); @Override Map<TaskId,Task> getTasks(TaskType taskType); @Override JobState getState(); @Override /** * The only entry point to change the Job. */ void handle(JobEvent event); @Override String getUserName(); @Override String getName(); @Override int getTotalMaps(); @Override int getTotalReduces(); public Configuration conf; public Path remoteJobConfFile; }### Answer:
@Test public void testCheckJobCompleteSuccess() { JobImpl mockJob = mock(JobImpl.class); OutputCommitter mockCommitter = mock(OutputCommitter.class); EventHandler mockEventHandler = mock(EventHandler.class); JobContext mockJobContext = mock(JobContext.class); when(mockJob.getCommitter()).thenReturn(mockCommitter); when(mockJob.getEventHandler()).thenReturn(mockEventHandler); when(mockJob.getJobContext()).thenReturn(mockJobContext); doNothing().when(mockJob).setFinishTime(); doNothing().when(mockJob).logJobHistoryFinishedEvent(); when(mockJob.finished(any(JobState.class))).thenReturn(JobState.SUCCEEDED); try { doNothing().when(mockCommitter).commitJob(any(JobContext.class)); } catch (IOException e) { } doNothing().when(mockEventHandler).handle(any(JobHistoryEvent.class)); Assert.assertNotNull("checkJobCompleteSuccess incorrectly returns null " + "for successful job", JobImpl.checkJobCompleteSuccess(mockJob)); Assert.assertEquals("checkJobCompleteSuccess returns incorrect state", JobImpl.checkJobCompleteSuccess(mockJob), JobState.SUCCEEDED); }