JobSupervisor implements ComponentLifeCycle { void cleanOldJobs() { try { final Iterator<Long> startedJobs = jobDaoProvider.get().getAllJobIds(JobStatus.STARTED); int stoppedJobs = 0; while (startedJobs.hasNext()) { long id = startedJobs.next(); Job job = jobDaoProvider.get().read(id); long timeDiff = jobTimeoutTime * TimeUtils.SECOND_IN_MILLIS; Date endTime = new Date(); endTime.setTime(job.getActualStart().getTime() + timeDiff); if (new Date().after(endTime)) { final String msg = " Job " + id + " has exceeded its timeout of " + (jobTimeoutTime / TimeUtils.HOUR_IN_MINUTES) + " minutes." + " Changing status to " + "FAILED."; log.warn(msg); job.setStatus(JobStatus.FAILED); job.appendHarvestErrors(msg); jobDaoProvider.get().update(job); ++stoppedJobs; } } if (stoppedJobs > 0) { log.warn("Changed {} jobs from STARTED to FAILED", stoppedJobs); } } catch (Throwable t) { log.error("Unable to stop obsolete jobs", t); } } JobSupervisor(Provider<JobDAO> jobDaoProvider, Long jobTimeoutTime); @Override void start(); @Override void shutdown();  }
@Test public void testCleanOldJobsMultipleJobs() { Long jobTimeoutTime = 1L; jobSupervisor = new JobSupervisor(jobDAOProvider, jobTimeoutTime); List<Long> jobIDs = Arrays.asList(1L, 2L, 3L); when(jobDaoMock.getAllJobIds(JobStatus.STARTED)).thenReturn(jobIDs.iterator()); Job pastObsoleteJobMock = mock(Job.class); Job pastActiveMock = mock(Job.class); Job futureActiveMock = mock(Job.class); when(jobDaoMock.read(1L)).thenReturn(pastObsoleteJobMock); when(jobDaoMock.read(2L)).thenReturn(pastActiveMock); when(jobDaoMock.read(3L)).thenReturn(futureActiveMock); Date inTheObsoletePast = new Date(System.currentTimeMillis() - 10000); Date inTheActivePast = new Date(System.currentTimeMillis() - 1); Date inTheActiveFuture = new Date(System.currentTimeMillis() + 10000); when(pastObsoleteJobMock.getActualStart()).thenReturn(inTheObsoletePast); when(pastActiveMock.getActualStart()).thenReturn(inTheActivePast); when(futureActiveMock.getActualStart()).thenReturn(inTheActiveFuture); jobSupervisor.cleanOldJobs(); verify(jobDaoMock).getAllJobIds(JobStatus.STARTED); verify(jobDaoMock).read(jobIDs.get(0)); verify(jobDaoMock).read(jobIDs.get(1)); verify(jobDaoMock).read(jobIDs.get(2)); verify(pastObsoleteJobMock).getActualStart(); verify(pastActiveMock).getActualStart(); verify(futureActiveMock).getActualStart(); verify(pastObsoleteJobMock).setStatus(JobStatus.FAILED); verify(pastObsoleteJobMock).appendHarvestErrors(Matchers.any(String.class)); verifyNoMoreInteractions(pastActiveMock, futureActiveMock); verify(jobDaoMock).update(pastObsoleteJobMock); verifyNoMoreInteractions(jobDaoMock); }
@Test public void testCleanOldJobsNoJobs() { Long jobTimeoutTime = 1L; jobSupervisor = new JobSupervisor(jobDAOProvider, jobTimeoutTime); List<Long> jobIDs = Arrays.asList(new Long[] {}); when(jobDaoMock.getAllJobIds(JobStatus.STARTED)).thenReturn(jobIDs.iterator()); jobSupervisor.cleanOldJobs(); verify(jobDaoMock).getAllJobIds(JobStatus.STARTED); verifyNoMoreInteractions(jobDaoMock); }