Java 类org.springframework.batch.core.BatchStatus 实例源码

@Bean
public Job discreteJob() {
    AbstractJob job = new AbstractJob("discreteRegisteredJob") {

        @Override
        public Collection<String> getStepNames() {
            return Collections.emptySet();
        }

        @Override
        public Step getStep(String stepName) {
            return null;
        }

        @Override
        protected void doExecute(JobExecution execution)
                throws JobExecutionException {
            execution.setStatus(BatchStatus.COMPLETED);
        }
    };
    job.setJobRepository(this.jobRepository);
    return job;
}

@Bean
public Job discreteJob() {
    AbstractJob job = new AbstractJob("discreteLocalJob") {

        @Override
        public Collection<String> getStepNames() {
            return Collections.emptySet();
        }

        @Override
        public Step getStep(String stepName) {
            return null;
        }

        @Override
        protected void doExecute(JobExecution execution)
                throws JobExecutionException {
            execution.setStatus(BatchStatus.COMPLETED);
        }
    };
    job.setJobRepository(this.jobRepository);
    return job;
}

@Bean
public Job discreteJob() {
    AbstractJob job = new AbstractJob("discreteRegisteredJob") {

        @Override
        public Collection<String> getStepNames() {
            return Collections.emptySet();
        }

        @Override
        public Step getStep(String stepName) {
            return null;
        }

        @Override
        protected void doExecute(JobExecution execution)
                throws JobExecutionException {
            execution.setStatus(BatchStatus.COMPLETED);
        }
    };
    job.setJobRepository(this.jobRepository);
    return job;
}

@Bean
public Job discreteJob() {
    AbstractJob job = new AbstractJob("discreteLocalJob") {

        @Override
        public Collection<String> getStepNames() {
            return Collections.emptySet();
        }

        @Override
        public Step getStep(String stepName) {
            return null;
        }

        @Override
        protected void doExecute(JobExecution execution)
                throws JobExecutionException {
            execution.setStatus(BatchStatus.COMPLETED);
        }
    };
    job.setJobRepository(this.jobRepository);
    return job;
}

@Before
public void setup() throws Exception {
    if (!initialized) {
        registerApp(task, "timestamp");
        initialize();
        createJobExecution(JOB_NAME, BatchStatus.STARTED);

        documentation.dontDocument(() -> this.mockMvc.perform(
                post("/tasks/definitions")
                        .param("name", "DOCJOB_1")
                        .param("definition", "timestamp --format='YYYY MM DD'"))
                .andExpect(status().isOk()));

        initialized = true;
    }
}

@Before
public void setup() throws Exception {
    if (!initialized) {
        registerApp(task, "timestamp");
        initialize();
        createJobExecution(JOB_NAME, BatchStatus.STARTED);
        createJobExecution(JOB_NAME + "_1", BatchStatus.STOPPED);

        documentation.dontDocument(() -> this.mockMvc.perform(
                post("/tasks/definitions")
                        .param("name", "DOCJOB_1")
                        .param("definition", "timestamp --format='YYYY MM DD'"))
                .andExpect(status().isOk()));

        initialized = true;
    }
}

@Bean
public Job discreteJob() {
    AbstractJob job = new AbstractJob("discreteRegisteredJob") {

        @Override
        public Collection<String> getStepNames() {
            return Collections.emptySet();
        }

        @Override
        public Step getStep(String stepName) {
            return null;
        }

        @Override
        protected void doExecute(JobExecution execution)
                throws JobExecutionException {
            execution.setStatus(BatchStatus.COMPLETED);
        }
    };
    job.setJobRepository(this.jobRepository);
    return job;
}

@Bean
public Job discreteJob() {
    AbstractJob job = new AbstractJob("discreteLocalJob") {

        @Override
        public Collection<String> getStepNames() {
            return Collections.emptySet();
        }

        @Override
        public Step getStep(String stepName) {
            return null;
        }

        @Override
        protected void doExecute(JobExecution execution)
                throws JobExecutionException {
            execution.setStatus(BatchStatus.COMPLETED);
        }
    };
    job.setJobRepository(this.jobRepository);
    return job;
}

/**
 * @param resultSet Set the result set
 * @param rowNumber Set the row number
 * @throws SQLException if there is a problem
 * @return a job execution instance
 */
public final JobExecution mapRow(final ResultSet resultSet,
        final int rowNumber) throws SQLException {
    JobInstance jobInstance = new JobInstance(resultSet.getBigDecimal(
            "JOB_INSTANCE_ID").longValue(),
            new JobParameters(), resultSet.getString("JOB_NAME"));
    JobExecution jobExecution = new JobExecution(jobInstance,
            resultSet.getBigDecimal("JOB_EXECUTION_ID").longValue());
    jobExecution.setStartTime(resultSet.getTimestamp("START_TIME"));
    jobExecution.setCreateTime(resultSet.getTimestamp("CREATE_TIME"));
    jobExecution.setEndTime(resultSet.getTimestamp("END_TIME"));
    jobExecution.setStatus(BatchStatus.valueOf(resultSet
            .getString("STATUS")));
    ExitStatus exitStatus = new ExitStatus(
            resultSet.getString("EXIT_CODE"),
            resultSet.getString("EXIT_MESSAGE"));
    jobExecution.setExitStatus(exitStatus);
    return jobExecution;
}

@Override
public List<Resource> listResourcesToHarvest(Integer limit, DateTime now, String fetch) {
    Criteria criteria = getSession().createCriteria(type);
    criteria.add(Restrictions.isNotNull("resourceType"));
    criteria.add(Restrictions.in("status", Arrays.asList(new BatchStatus[] {BatchStatus.COMPLETED, BatchStatus.FAILED,BatchStatus.ABANDONED, BatchStatus.STOPPED})));
    criteria.add(Restrictions.eq("scheduled", Boolean.TRUE));
    criteria.add(Restrictions.disjunction().add(Restrictions.lt("nextAvailableDate", now)).add(Restrictions.isNull("nextAvailableDate")));

    if (limit != null) {
        criteria.setMaxResults(limit);
    }
    enableProfilePreQuery(criteria, fetch);
    criteria.addOrder( Property.forName("nextAvailableDate").asc() );
    List<Resource> result = (List<Resource>) criteria.list();
    for(Resource t : result) {
        enableProfilePostQuery(t, fetch);
    }
    return result;
}

/**
 * @param status
 *            Set the status
 * @return true if the job is startable
 */
public static Boolean isStartable(BatchStatus status) {
 if (status == null) {
     return Boolean.TRUE;
 } else {
     switch (status) {
     case STARTED:
     case STARTING:
     case STOPPING:
     case UNKNOWN:
         return Boolean.FALSE;
     case COMPLETED:
     case FAILED:
     case STOPPED:
     default:
         return Boolean.TRUE;
     }
 }
}

@Override
public void notify(JobExecutionException jobExecutionException, String resourceIdentifier) {
    if(resourceIdentifier != null) {
        Resource resource = service.find(resourceIdentifier,"job-with-source");
        resource.setJobId(null);
        resource.setDuration(null);
        resource.setExitCode("FAILED");
        resource.setExitDescription(jobExecutionException.getLocalizedMessage());
        resource.setJobInstance(null);
        resource.setResource(null);
        resource.setStartTime(null);
        resource.setStatus(BatchStatus.FAILED);
        resource.setProcessSkip(0);
        resource.setRecordsRead(0);
        resource.setReadSkip(0);
        resource.setWriteSkip(0);
        resource.setWritten(0);

        service.saveOrUpdate(resource);
        solrIndexingListener.indexObject(resource);
    }

}

/**
 * Update and retrieve job execution - check attributes have changed as
 * expected.
 */
@Transactional
@Test
public void testUpdateExecution() {
    execution.setStatus(BatchStatus.STARTED);
    jobExecutionDao.saveJobExecution(execution);

    execution.setLastUpdated(new Date(0));
    execution.setStatus(BatchStatus.COMPLETED);
    jobExecutionDao.updateJobExecution(execution);

    JobExecution updated = jobExecutionDao.findJobExecutions(jobInstance).get(0);
    assertEquals(execution, updated);
    assertEquals(BatchStatus.COMPLETED, updated.getStatus());
    assertExecutionsAreEqual(execution, updated);
}

/**
 * Successful synchronization from STARTED to STOPPING status.
 */
@Transactional
@Test
public void testSynchronizeStatusUpgrade() {

    JobExecution exec1 = new JobExecution(jobInstance, jobParameters);
    exec1.setStatus(BatchStatus.STOPPING);
    jobExecutionDao.saveJobExecution(exec1);

    JobExecution exec2 = new JobExecution(jobInstance, jobParameters);
    assertTrue(exec1.getId() != null);
    exec2.setId(exec1.getId());

    exec2.setStatus(BatchStatus.STARTED);
    exec2.setVersion(7);
    assertTrue(exec1.getVersion() != exec2.getVersion());
    assertTrue(exec1.getStatus() != exec2.getStatus());

    jobExecutionDao.synchronizeStatus(exec2);

    assertEquals(exec1.getVersion(), exec2.getVersion());
    assertEquals(exec1.getStatus(), exec2.getStatus());

}

/**
 * UNKNOWN status won't be changed by synchronizeStatus, because it is the
 * 'largest' BatchStatus (will not downgrade).
 */
@Transactional
@Test
public void testSynchronizeStatusDowngrade() {

    JobExecution exec1 = new JobExecution(jobInstance, jobParameters);
    exec1.setStatus(BatchStatus.STARTED);
    jobExecutionDao.saveJobExecution(exec1);

    JobExecution exec2 = new JobExecution(jobInstance, jobParameters);
    Assert.state(exec1.getId() != null);
    exec2.setId(exec1.getId());

    exec2.setStatus(BatchStatus.UNKNOWN);
    exec2.setVersion(7);
    Assert.state(exec1.getVersion() != exec2.getVersion());
    Assert.state(exec1.getStatus().isLessThan(exec2.getStatus()));

    jobExecutionDao.synchronizeStatus(exec2);

    assertEquals(exec1.getVersion(), exec2.getVersion());
    assertEquals(BatchStatus.UNKNOWN, exec2.getStatus());
}

@Override
public List<Resource> listResourcesToHarvest(Integer limit, DateTime now, String fetch) {
    Criteria criteria = getSession().createCriteria(type);
    criteria.add(Restrictions.isNotNull("resourceType"));
    criteria.add(Restrictions.in("status", Arrays.asList(new BatchStatus[] {BatchStatus.COMPLETED, BatchStatus.FAILED,BatchStatus.ABANDONED, BatchStatus.STOPPED})));
    criteria.add(Restrictions.eq("scheduled", Boolean.TRUE));
    criteria.add(Restrictions.disjunction().add(Restrictions.lt("nextAvailableDate", now)).add(Restrictions.isNull("nextAvailableDate")));

    if (limit != null) {
        criteria.setMaxResults(limit);
    }
    enableProfilePreQuery(criteria, fetch);
    criteria.addOrder( Property.forName("nextAvailableDate").asc() );
    List<Resource> result = (List<Resource>) criteria.list();
    for(Resource t : result) {
        enableProfilePostQuery(t, fetch);
    }
    return result;
}

private boolean harvestSuccessful(JobConfiguration job) {
    DateTime start = DateTime.now();

    while(new Period(start, DateTime.now()).getSeconds() < 20) {
        jobConfigurationService.refresh(job);
        if(BatchStatus.COMPLETED.equals(job.getJobStatus())) {
            logger.info("Succesfully completed {}", job.getDescription());
            return true;
        }

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    return false;
}

@Test
public void testJobInterruptedException() throws Exception {
    StepExecution workerStep = new StepExecution("workerStep", new JobExecution(1L), 2L);

    when(this.environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID)).thenReturn(true);
    when(this.environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID)).thenReturn(true);
    when(this.environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)).thenReturn(true);
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)).thenReturn("workerStep");
    when(this.beanFactory.getBeanNamesForType(Step.class)).thenReturn(new String[] {"workerStep", "foo", "bar"});
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID)).thenReturn("2");
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID)).thenReturn("1");
    when(this.jobExplorer.getStepExecution(1L, 2L)).thenReturn(workerStep);
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)).thenReturn("workerStep");
    when(this.beanFactory.getBean("workerStep", Step.class)).thenReturn(this.step);
    doThrow(new JobInterruptedException("expected")).when(this.step).execute(workerStep);

    handler.run();

    verify(this.jobRepository).update(this.stepExecutionArgumentCaptor.capture());

    assertEquals(BatchStatus.STOPPED, this.stepExecutionArgumentCaptor.getValue().getStatus());
}

@Test
public void testRuntimeException() throws Exception {
    StepExecution workerStep = new StepExecution("workerStep", new JobExecution(1L), 2L);

    when(this.environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID)).thenReturn(true);
    when(this.environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID)).thenReturn(true);
    when(this.environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)).thenReturn(true);
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)).thenReturn("workerStep");
    when(this.beanFactory.getBeanNamesForType(Step.class)).thenReturn(new String[] {"workerStep", "foo", "bar"});
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID)).thenReturn("2");
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID)).thenReturn("1");
    when(this.jobExplorer.getStepExecution(1L, 2L)).thenReturn(workerStep);
    when(this.environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)).thenReturn("workerStep");
    when(this.beanFactory.getBean("workerStep", Step.class)).thenReturn(this.step);
    doThrow(new RuntimeException("expected")).when(this.step).execute(workerStep);

    handler.run();

    verify(this.jobRepository).update(this.stepExecutionArgumentCaptor.capture());

    assertEquals(BatchStatus.FAILED, this.stepExecutionArgumentCaptor.getValue().getStatus());
}

private JobParameters createJobParametersWithIncrementerIfAvailable(String parameters, Job job) throws JobParametersNotFoundException {
    JobParameters jobParameters = jobParametersConverter.getJobParameters(PropertiesConverter.stringToProperties(parameters));
    // use JobParametersIncrementer to create JobParameters if incrementer is set and only if the job is no restart
    if (job.getJobParametersIncrementer() != null){
        JobExecution lastJobExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
        boolean restart = false;
        // check if job failed before
        if (lastJobExecution != null) {
            BatchStatus status = lastJobExecution.getStatus();
            if (status.isUnsuccessful() && status != BatchStatus.ABANDONED) {
                restart = true;
            }
        }
        // if it's not a restart, create new JobParameters with the incrementer
        if (!restart) {
            JobParameters nextParameters = getNextJobParameters(job);
            Map<String, JobParameter> map = new HashMap<String, JobParameter>(nextParameters.getParameters());
            map.putAll(jobParameters.getParameters());
            jobParameters = new JobParameters(map);
        }
    }
    return jobParameters;
}

@Test
public void testRunJob() throws InterruptedException {
    Long executionId = restTemplate.postForObject("http://localhost:" + port + "/batch/operations/jobs/flatFile2JobXml", "", Long.class);
    while (!restTemplate.getForObject("http://localhost:" + port + "/batch/operations/jobs/executions/{executionId}", String.class, executionId)
            .equals("COMPLETED")) {
        Thread.sleep(1000);
    }
    String log = restTemplate.getForObject("http://localhost:" + port + "/batch/operations/jobs/executions/{executionId}/log", String.class,
            executionId);
    assertThat(log.length() > 20, is(true));
    JobExecution jobExecution = jobExplorer.getJobExecution(executionId);
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    String jobExecutionString = restTemplate.getForObject("http://localhost:" + port + "/batch/monitoring/jobs/executions/{executionId}",
            String.class, executionId);
    assertThat(jobExecutionString.contains("COMPLETED"), is(true));
}

@Test
public void testRunFlatFileToDbNoSkipJob_Success() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbNoSkipJob", "metrics/flatFileToDbNoSkipJob_Success.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 5L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(2L)
            .withStreamOpenCount(1L).withStreamUpdateCount(3L).withStreamCloseCount(0L).withBeforeReadCount(6L).withReadCount(6L)
            .withAfterReadCount(5L).withReadErrorCount(0L).withBeforeProcessCount(5L).withProcessCount(5L).withAfterProcessCount(5L)
            .withProcessErrorCount(0L).withBeforeWriteCount(5L).withWriteCount(writeCount).withAfterWriteCount(5L).withAfterChunkCount(2L)
            .withChunkErrorCount(0L).withSkipInReadCount(0L).withSkipInProcessCount(0L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader.findOne("gauge.batch.flatFileToDbNoSkipJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue())); // TODO
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbNoSkipJob_Failed() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbNoSkipJob", "metrics/flatFileToDbNoSkipJob_Failed.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.FAILED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 3L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(1L)
            .withStreamOpenCount(1L).withStreamUpdateCount(2L).withStreamCloseCount(0L).withBeforeReadCount(6L).withReadCount(6L)
            .withAfterReadCount(5L).withReadErrorCount(0L).withBeforeProcessCount(3L).withProcessCount(3L).withAfterProcessCount(3L)
            .withProcessErrorCount(1L).withBeforeWriteCount(3L).withWriteCount(writeCount).withAfterWriteCount(3L).withAfterChunkCount(1L)
            .withChunkErrorCount(1L).withSkipInReadCount(0L).withSkipInProcessCount(0L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader.findOne("gauge.batch.flatFileToDbNoSkipJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInProcess() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipJob", "metrics/flatFileToDbSkipJob_SkipInProcess.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 7L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(3L)
            .withStreamOpenCount(1L).withStreamUpdateCount(4L).withStreamCloseCount(0L).withBeforeReadCount(9L).withReadCount(9L)
            .withAfterReadCount(8L).withReadErrorCount(0L).withBeforeProcessCount(7L).withProcessCount(7L).withAfterProcessCount(7L)
            .withProcessErrorCount(1L).withBeforeWriteCount(7L).withWriteCount(writeCount).withAfterWriteCount(7L).withAfterChunkCount(3L)
            .withChunkErrorCount(1L).withSkipInReadCount(0L).withSkipInProcessCount(1L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader.findOne("gauge.batch.flatFileToDbSkipJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInProcess_Failed() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipJob", "metrics/flatFileToDbSkipJob_SkipInProcess_Failed.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.FAILED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 7L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(3L)
            .withStreamOpenCount(1L).withStreamUpdateCount(4L).withStreamCloseCount(0L).withBeforeReadCount(12L).withReadCount(12L)
            .withAfterReadCount(12L).withReadErrorCount(0L).withBeforeProcessCount(7L).withProcessCount(7L).withAfterProcessCount(7L)
            .withProcessErrorCount(5L).withBeforeWriteCount(7L).withWriteCount(writeCount).withAfterWriteCount(7L).withAfterChunkCount(3L)
            .withChunkErrorCount(6L).withSkipInReadCount(0L).withSkipInProcessCount(2L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader.findOne("gauge.batch.flatFileToDbSkipJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInWrite() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipJob", "metrics/flatFileToDbSkipJob_SkipInWrite.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 7L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(4L)
            .withStreamOpenCount(1L).withStreamUpdateCount(4L).withStreamCloseCount(0L).withBeforeReadCount(9L).withReadCount(9L)
            .withAfterReadCount(8L).withReadErrorCount(0L).withBeforeProcessCount(7L).withProcessCount(7L).withAfterProcessCount(7L)
            .withProcessErrorCount(0L).withBeforeWriteCount(5L).withWriteCount(writeCount).withAfterWriteCount(7L).withWriteErrorCount(4L)
            .withAfterChunkCount(4L).withChunkErrorCount(2L).withSkipInReadCount(0L).withSkipInProcessCount(0L).withSkipInWriteCount(1L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader.findOne("gauge.batch.flatFileToDbSkipJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInRead() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipJob", "metrics/flatFileToDbSkipJob_SkipInRead.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 7L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(3L)
            .withStreamOpenCount(1L).withStreamUpdateCount(4L).withStreamCloseCount(0L).withBeforeReadCount(9L).withReadCount(9L)
            .withAfterReadCount(7L).withReadErrorCount(1L).withBeforeProcessCount(7L).withProcessCount(7L).withAfterProcessCount(7L)
            .withProcessErrorCount(0L).withBeforeWriteCount(7L).withWriteCount(writeCount).withAfterWriteCount(7L).withWriteErrorCount(0L)
            .withAfterChunkCount(3L).withChunkErrorCount(0L).withSkipInReadCount(1L).withSkipInProcessCount(0L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader.findOne("gauge.batch.flatFileToDbSkipJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInProcess_ProcessorNonTransactional() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipProcessorNonTransactionalJob", "metrics/flatFileToDbSkipJob_SkipInProcess.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 7L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(3L)
            .withStreamOpenCount(1L).withStreamUpdateCount(4L).withStreamCloseCount(0L).withBeforeReadCount(9L).withReadCount(9L)
            .withAfterReadCount(8L).withReadErrorCount(0L).withBeforeProcessCount(8L).withProcessCount(8L).withAfterProcessCount(7L)
            .withProcessErrorCount(1L).withBeforeWriteCount(7L).withWriteCount(writeCount).withAfterWriteCount(7L).withAfterChunkCount(3L)
            .withChunkErrorCount(1L).withSkipInReadCount(0L).withSkipInProcessCount(1L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat(
            (Double) metricReader.findOne(
                    "gauge.batch.flatFileToDbSkipProcessorNonTransactionalJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInWrite_ProcessorNonTransactional() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipProcessorNonTransactionalJob", "metrics/flatFileToDbSkipJob_SkipInWrite.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 7L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(4L)
            .withStreamOpenCount(1L).withStreamUpdateCount(4L).withStreamCloseCount(0L).withBeforeReadCount(9L).withReadCount(9L)
            .withAfterReadCount(8L).withReadErrorCount(0L).withBeforeProcessCount(8L).withProcessCount(8L).withAfterProcessCount(8L)
            .withProcessErrorCount(0L).withBeforeWriteCount(5L).withWriteCount(writeCount).withAfterWriteCount(7L).withWriteErrorCount(4L)
            .withAfterChunkCount(4L).withChunkErrorCount(2L).withSkipInReadCount(0L).withSkipInProcessCount(0L).withSkipInWriteCount(1L).build();
    // TODO Bug in beforeWrite listener in Spring Batch?
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat(
            (Double) metricReader.findOne(
                    "gauge.batch.flatFileToDbSkipProcessorNonTransactionalJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunFlatFileToDbSkipJob_SkipInProcess_ReaderTransactional() throws InterruptedException {
    JobExecution jobExecution = runJob("flatFileToDbSkipReaderTransactionalJob", "metrics/flatFileToDbSkipJob_SkipInProcess.csv");
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    ExecutionContext executionContext = jobExecution.getStepExecutions().iterator().next().getExecutionContext();
    long writeCount = 5L;
    MetricValidator validator = MetricValidatorBuilder.metricValidator().withExecutionContext(executionContext).withBeforeChunkCount(2L)
            .withStreamOpenCount(1L).withStreamUpdateCount(3L).withStreamCloseCount(0L).withBeforeReadCount(6L).withReadCount(6L)
            .withAfterReadCount(5L).withReadErrorCount(0L).withBeforeProcessCount(5L).withProcessCount(5L).withAfterProcessCount(5L)
            .withProcessErrorCount(1L).withBeforeWriteCount(5L).withWriteCount(writeCount).withAfterWriteCount(5L).withAfterChunkCount(2L)
            .withChunkErrorCount(1L).withSkipInReadCount(0L).withSkipInProcessCount(0L).withSkipInWriteCount(0L).build();
    validator.validate();
    // if one is correct, all will be in the metricReader, so I check just one
    assertThat((Double) metricReader
            .findOne("gauge.batch.flatFileToDbSkipReaderTransactionalJob.step." + MetricNames.PROCESS_COUNT.getName()).getValue(),
            is(notNullValue()));
    assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ITEM", Long.class), is(writeCount));
}

@Test
public void testRunJob() throws InterruptedException {
    Long executionId = restTemplate.postForObject("http://localhost:" + port + "/batch/operations/jobs/simpleBatchMetricsJob", "", Long.class);
    while (!restTemplate.getForObject("http://localhost:" + port + "/batch/operations/jobs/executions/{executionId}", String.class, executionId)
            .equals("COMPLETED")) {
        Thread.sleep(1000);
    }
    String log = restTemplate.getForObject("http://localhost:" + port + "/batch/operations/jobs/executions/{executionId}/log", String.class,
            executionId);
    assertThat(log.length() > 20, is(true));
    JobExecution jobExecution = jobExplorer.getJobExecution(executionId);
    assertThat(jobExecution.getStatus(), is(BatchStatus.COMPLETED));
    String jobExecutionString = restTemplate.getForObject("http://localhost:" + port + "/batch/monitoring/jobs/executions/{executionId}",
            String.class, executionId);
    assertThat(jobExecutionString.contains("COMPLETED"), is(true));
    Metric<?> metric = metricReader.findOne("gauge.batch.simpleBatchMetricsJob.simpleBatchMetricsStep.processor");
    assertThat(metric, is(notNullValue()));
    assertThat((Double) metric.getValue(), is(notNullValue()));
    assertThat((Double) metric.getValue(), is(7.0));
}

@Test
public void testRunJob() throws InterruptedException{
    Long executionId = restTemplate.postForObject("http://localhost:"+port+"/batch/operations/jobs/delayJob", "",Long.class);
    Thread.sleep(500);
    String runningExecutions = restTemplate.getForObject("http://localhost:"+port+"/batch/monitoring/jobs/runningexecutions", String.class);
    assertThat(runningExecutions.contains(executionId.toString()),is(true));
    String runningExecutionsForDelayJob = restTemplate.getForObject("http://localhost:"+port+"/batch/monitoring/jobs/runningexecutions/delayJob", String.class);
    assertThat(runningExecutionsForDelayJob.contains(executionId.toString()),is(true));
    restTemplate.delete("http://localhost:"+port+"/batch/operations/jobs/executions/{executionId}",executionId);
    Thread.sleep(1500);

    JobExecution jobExecution = jobExplorer.getJobExecution(executionId);
    assertThat(jobExecution.getStatus(),is(BatchStatus.STOPPED));
    String jobExecutionString = restTemplate.getForObject("http://localhost:"+port+"/batch/monitoring/jobs/executions/{executionId}",String.class,executionId);
    assertThat(jobExecutionString.contains("STOPPED"),is(true));
}

@Test
public void testXml() throws Exception {
    JobExecution exec = jobLauncherTestUtils.launchJob();
    Assertions.assertThat(exec.getStatus()).isEqualTo(BatchStatus.COMPLETED);

    Resource ouput = new FileSystemResource(JobXmlMultiFileConfiguration.OUTPUT_FILE + ".1");
    String content = IOUtils.toString(ouput.getInputStream());
    assertXpathEvaluatesTo("1000", "count(//product)", content);

    ouput = new FileSystemResource(JobXmlMultiFileConfiguration.OUTPUT_FILE + ".2");
    content = IOUtils.toString(ouput.getInputStream());
    assertXpathEvaluatesTo("1000", "count(//product)", content);

    ouput = new FileSystemResource(JobXmlMultiFileConfiguration.OUTPUT_FILE + ".3");
    content = IOUtils.toString(ouput.getInputStream());
    assertXpathEvaluatesTo("1000", "count(//product)", content);

    ouput = new FileSystemResource(JobXmlMultiFileConfiguration.OUTPUT_FILE + ".4");
    content = IOUtils.toString(ouput.getInputStream());
    assertXpathEvaluatesTo("1000", "count(//product)", content);

    ouput = new FileSystemResource(JobXmlMultiFileConfiguration.OUTPUT_FILE + ".5");
    content = IOUtils.toString(ouput.getInputStream());
    assertXpathEvaluatesTo("517", "count(//product)", content);
}

/**
 * JobOperator 가 Job 중단을 수행합니다.
 *
 * @throws Exception
 */
@Test
public void stopWithJobOperator() throws Exception {
    JobExecution jobExecution = jobLauncher.run(jobOperatorJob, new JobParameters());
    assertThat(jobExecution.getStatus()).isIn(BatchStatus.STARTING, BatchStatus.STARTED);
    Set<Long> runningExecutions = jobOperator.getRunningExecutions(jobOperatorJob.getName());
    assertThat(runningExecutions.size()).isEqualTo(1);

    Long executionId = runningExecutions.iterator().next();
    boolean stopMessageSent = jobOperator.stop(executionId);
    assertThat(stopMessageSent).isTrue();

    waitForTermination(jobOperatorJob);
    runningExecutions = jobOperator.getRunningExecutions(jobOperatorJob.getName());
    assertThat(runningExecutions.size()).isEqualTo(0);
}

@Test
public void sunnyDay() throws Exception {
    int read = 12;
    configureServiceForRead(service, read);
    JobExecution exec = jobLauncher.run(
            job,
            new JobParametersBuilder().addLong("time", System.currentTimeMillis())
                                      .toJobParameters());
    assertThat(exec.getStatus()).isEqualTo(BatchStatus.COMPLETED);
    assertRead(read, exec);
    assertWrite(read, exec);
    assertReadSkip(0, exec);
    assertProcessSkip(0, exec);
    assertWriteSkip(0, exec);
    assertCommit(3, exec);
    assertRollback(0, exec);
}

@Router
public List<String> routeJobExecution(JobExecution jobExecution) {

    final List<String> routeToChannels = new ArrayList<String>();

    if (jobExecution.getStatus().equals(BatchStatus.FAILED)) {
        routeToChannels.add("jobRestarts");
    }
    else {

        if (jobExecution.getStatus().equals(BatchStatus.COMPLETED)) {
            routeToChannels.add("completeApplication");
        }

        routeToChannels.add("notifiableExecutions");
    }

    return routeToChannels;
}

@Router
public List<String> routeJobExecution(JobExecution jobExecution) {

    final List<String> routeToChannels = new ArrayList<String>();

    if (jobExecution.getStatus().equals(BatchStatus.FAILED)) {
        routeToChannels.add("jobRestarts");
    }
    else {

        if (jobExecution.getStatus().equals(BatchStatus.COMPLETED)) {
            routeToChannels.add("completeApplication");
        }

        routeToChannels.add("notifiableExecutions");
    }

    return routeToChannels;
}

@Router
public List<String> routeJobExecution(JobExecution jobExecution) {

    final List<String> routeToChannels = new ArrayList<String>();

    if (jobExecution.getStatus().equals(BatchStatus.FAILED)) {
        routeToChannels.add("jobRestarts");
    }
    else {

        if (jobExecution.getStatus().equals(BatchStatus.COMPLETED)) {
            routeToChannels.add("completeApplication");
        }

        routeToChannels.add("notifiableExecutions");
    }

    return routeToChannels;
}

@Router
public List<String> routeJobExecution(JobExecution jobExecution) {

    final List<String> routeToChannels = new ArrayList<String>();

    if (jobExecution.getStatus().equals(BatchStatus.FAILED)) {
        routeToChannels.add("jobRestarts");
    }
    else {

        if (jobExecution.getStatus().equals(BatchStatus.COMPLETED)) {
            routeToChannels.add("completeApplication");
        }

        routeToChannels.add("notifiableExecutions");
    }

    return routeToChannels;
}

@Override
public void afterJob(JobExecution jobExecution) {
    if(jobExecution.getStatus() == BatchStatus.COMPLETED) {
        logger.info("!!! JOB FINISHED! LAST POSTID IMPORTED: " +
                jobExecution.getExecutionContext().get("postId") );
    }
}