JobSupervisor implements ComponentLifeCycle { void rescheduleLeftOverJobs() { final Iterator<Long> jobs = jobDaoProvider.get().getAllJobIds(JobStatus.SUBMITTED); int resubmitcount = 0; while (jobs.hasNext()) { long oldID = jobs.next(); long newID = jobDaoProvider.get().rescheduleJob(oldID); log.info("Resubmitting old job {} as {}", oldID, newID); ++resubmitcount; } log.info("{} jobs has been resubmitted.", resubmitcount); } JobSupervisor(Provider<JobDAO> jobDaoProvider, Long jobTimeoutTime); @Override void start(); @Override void shutdown();  }
@Test public void testRescheduleMultipleSubmittedJobs() { Long jobTimeoutTime = 1L; jobSupervisor = new JobSupervisor(jobDAOProvider, jobTimeoutTime); List<Long> jobIDs = Arrays.asList(1L, 3L); when(jobDaoMock.getAllJobIds(JobStatus.SUBMITTED)).thenReturn(jobIDs.iterator()); jobSupervisor.rescheduleLeftOverJobs(); verify(jobDaoMock).getAllJobIds(JobStatus.SUBMITTED); verify(jobDaoMock).rescheduleJob(jobIDs.get(0)); verify(jobDaoMock).rescheduleJob(jobIDs.get(1)); verifyNoMoreInteractions(jobDaoMock); }
@Test public void testRescheduleNoSubmittedJobs() { Long jobTimeoutTime = 1L; jobSupervisor = new JobSupervisor(jobDAOProvider, jobTimeoutTime); List<Long> jobIDs = Arrays.asList(new Long[] {}); when(jobDaoMock.getAllJobIds(JobStatus.SUBMITTED)).thenReturn(jobIDs.iterator()); jobSupervisor.rescheduleLeftOverJobs(); verify(jobDaoMock).getAllJobIds(JobStatus.SUBMITTED); verifyNoMoreInteractions(jobDaoMock); }