BatchStartAndRecoveryServiceImpl implements BatchStartAndRecoveryService, JobExplorer { @Override public List<JobExecution> markInconsistentJobsAsFailed() throws NoSuchJobException, NoSuchJobExecutionException, JobExecutionNotRunningException, InterruptedException, JobExecutionAlreadyRunningException, JobInstanceAlreadyCompleteException, JobRestartException, JobParametersInvalidException, NoSuchJobInstanceException { List<JobExecution> result = new ArrayList<>(); Collection<String> jobNames = jobRegistry.getJobNames(); for (String jobName : jobNames) { LOGGER.info("Searching to recover jobName={}...", jobName); Set<Long> jobExecutionIds = getRunningExecutions(jobName); for (Long jobExecutionId : jobExecutionIds) { LOGGER.info("Found a job already running for jobExecutionId={}.", jobExecutionId); JobExecution jobExecution = findExecutionById(jobExecutionId); jobExecution.setEndTime(new Date()); jobExecution.setStatus(BatchStatus.FAILED); jobExecution.setExitStatus(RECOVERY_STATUS); jobExecution.getStepExecutions() .stream() .filter(stepExecution -> stepExecution.getStatus().isRunning()) .forEach(stepExecution -> { stepExecution.setEndTime(new Date()); stepExecution.setStatus(BatchStatus.FAILED); stepExecution.setExitStatus(RECOVERY_STATUS); updateStepExecution(stepExecution); }); result.add(jobExecution); updateJobExecution(jobExecution); LOGGER.info("Mark job execution with Id={} as FAILED!", jobExecutionId); } } return result; } BatchStartAndRecoveryServiceImpl(JobInstanceDao jobInstanceDao,
                                            JobExecutionDao jobExecutionDao,
                                            StepExecutionDao stepExecutionDao,
                                            ExecutionContextDao executionContextDao
    ); @Override List<JobExecution> stopAllRunningJobs(); @Override List<JobExecution> markInconsistentJobsAsFailed(); Set<Long> getRunningExecutions(String jobName); @Override List<JobExecution> restartAllFailedJobs(); @Override List<JobExecution> restartAllStoppedJobs(); List<JobExecution> restartJobs(String jobName, BatchStatus status); @Override BatchStatus getStatus(Long jobInstanceId); List<JobInstance> getJobInstances(String jobName, int start, int count); @Override JobExecution start(String jobName, JobParameters jobParameters); JobExecution start(String jobName, JobParameters jobParameters, boolean newInstance); @Override JobExecution restart(Long jobInstanceId); JobExecution restart(String jobName, JobParameters parameters); JobExecution findExecutionById(Long jobExecutionId); @Override JobExecution startNewInstance(String jobName, JobParameters jobParameters); @Override JobExecution stop(Long jobExecutionId); @Override List<String> getJobNames(); @Override List<JobInstance> findJobInstancesByJobName(String jobName, int start, int count); @Override int getJobInstanceCount(String jobName); @Override JobExecution refresh(JobExecution jobExecution); @Override JobExecution abandon(Long jobExecutionId); @Override JobExecution getJobExecution(Long executionId); @Override StepExecution getStepExecution(Long jobExecutionId, Long stepExecutionId); @Override JobInstance getJobInstance(Long instanceId); @Override List<JobExecution> getJobExecutions(JobInstance jobInstance); @Override Set<JobExecution> findRunningJobExecutions(String jobName); void updateJobExecution(JobExecution jobExecution); void updateStepExecution(StepExecution stepExecution); void setJobRegistry(JobRegistry jobRegistry); void setJobLauncher(JobLauncher jobLauncher); void setJobInstanceDao(JobInstanceDao jobInstanceDao); void setJobExecutionDao(JobExecutionDao jobExecutionDao); void setStepExecutionDao(StepExecutionDao stepExecutionDao); void setExecutionContextDao(ExecutionContextDao executionContextDao);  }
@Test public void testMarkInconsistentJobsAsFailed() throws Exception { expectGetRunningExecutions(Collections.singleton(jobExecution)); expectFindExecutionById(); expectUpdateJobExecution(); expectUpdateStepExecution(); replayAll(); List<JobExecution> jobExecutions = service.markInconsistentJobsAsFailed(); Assert.assertEquals(1, jobExecutions.size()); Assert.assertEquals(BatchStatus.FAILED, jobExecutions.get(0).getStatus()); }