diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java index 5d3bc8c74..d545dfd4b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java @@ -361,6 +361,9 @@ public class SimpleJobOperator implements JobOperator, InitializingBean { // Indicate the execution should be stopped by setting it's status to // 'STOPPING'. It is assumed that // the step implementation will check this status at chunk boundaries. + BatchStatus status = jobExecution.getStatus(); + Assert.state(status == BatchStatus.STARTED || status == BatchStatus.STARTING, + "JobExecution must be running so that it can be stopped"); jobExecution.setStatus(BatchStatus.STOPPING); jobRepository.update(jobExecution); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java index b937978ae..e59c2bfa7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java @@ -255,9 +255,14 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements } public void synchronizeStatus(JobExecution jobExecution) { + int currentVersion = getJdbcTemplate().queryForInt(getQuery(CURRENT_VERSION_JOB_EXECUTION), + jobExecution.getId()); - String status = getJdbcTemplate().queryForObject(getQuery(GET_STATUS), String.class, jobExecution.getId()); - jobExecution.setStatus(BatchStatus.valueOf(status)); + if (currentVersion != jobExecution.getVersion().intValue()) { + String status = getJdbcTemplate().queryForObject(getQuery(GET_STATUS), String.class, jobExecution.getId()); + jobExecution.setStatus(BatchStatus.valueOf(status)); + jobExecution.setVersion(currentVersion); + } } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java index 214e735e7..4e4218ebb 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JobExecutionDao.java @@ -61,8 +61,8 @@ public interface JobExecutionDao { /** * Because it may be possible that the status of a JobExecution is updated - * while running, the following method while synchronize only the status - * field. + * while running, the following method while synchronize only the status and + * version fields. * * @param jobExecution to be updated. */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java index fe30e1df6..b0874810c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java @@ -73,7 +73,7 @@ public class MapJobExecutionDao implements JobExecutionDao { Assert.notNull(persistedExecution, "JobExecution must already be saved"); synchronized (jobExecution) { - if (!persistedExecution.getVersion().equals(jobExecution.getVersion())){ + if (!persistedExecution.getVersion().equals(jobExecution.getVersion())) { throw new OptimisticLockingFailureException("Attempt to update step execution id=" + id + " with wrong version (" + jobExecution.getVersion() + "), where current version is " + persistedExecution.getVersion()); @@ -128,6 +128,10 @@ public class MapJobExecutionDao implements JobExecutionDao { } public void synchronizeStatus(JobExecution jobExecution) { - // no-op + JobExecution saved = getJobExecution(jobExecution.getId()); + if (saved.getVersion().intValue() != jobExecution.getVersion().intValue()) { + jobExecution.setStatus(saved.getStatus()); + jobExecution.setVersion(saved.getVersion()); + } } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java index b27c253a9..c27ab8789 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java @@ -438,6 +438,7 @@ public class SimpleJobTests { job.setSteps(Arrays.asList(new Step[] { step1, step2 })); job.execute(jobExecution); + assertEquals(BatchStatus.STOPPED, jobExecution.getStatus()); assertEquals(1, jobExecution.getAllFailureExceptions().size()); Throwable expected = jobExecution.getAllFailureExceptions().get(0); assertTrue("Wrong exception " + expected, expected instanceof JobInterruptedException); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java index 59dae1241..c7bd3c4cd 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java @@ -25,6 +25,7 @@ import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.partition.PartitionHandler; @@ -34,19 +35,21 @@ import org.springframework.batch.core.repository.support.MapJobRepositoryFactory import org.springframework.batch.core.step.StepSupport; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; - /** * @author Dave Syer - * + * */ public class PartitionStepTests { - + private PartitionStep step = new PartitionStep(); + private Step remote = new StepSupport("remote"); + private JobRepository jobRepository; @Before public void setUp() throws Exception { + MapJobRepositoryFactoryBean.clear(); MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); factory.setTransactionManager(new ResourcelessTransactionManager()); jobRepository = (JobRepository) factory.getObject(); @@ -68,7 +71,8 @@ public class PartitionStepTests { } }); step.afterPropertiesSet(); - StepExecution stepExecution = new JobExecution(0L).createStepExecution("foo"); + JobExecution jobExecution = jobRepository.createJobExecution("vanillaJob", new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution("foo"); jobRepository.add(stepExecution); step.execute(stepExecution); // one master and two workers @@ -91,7 +95,8 @@ public class PartitionStepTests { } }); step.afterPropertiesSet(); - StepExecution stepExecution = new JobExecution(0L).createStepExecution("foo"); + JobExecution jobExecution = jobRepository.createJobExecution("vanillaJob", new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution("foo"); jobRepository.add(stepExecution); step.execute(stepExecution); // one master and two workers diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java index cb7e604c8..1ec7625f1 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java @@ -22,12 +22,13 @@ import org.springframework.batch.core.StepExecution; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.test.context.junit4.AbstractTransactionalJUnit4SpringContextTests; import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.Assert; public abstract class AbstractJobExecutionDaoTests extends AbstractTransactionalJUnit4SpringContextTests { protected JobExecutionDao dao; - protected JobInstance jobInstance = new JobInstance((long) 1, new JobParameters(), "execTestJob"); + protected JobInstance jobInstance = new JobInstance(1L, new JobParameters(), "execTestJob"); protected JobExecution execution = new JobExecution(jobInstance); @@ -248,7 +249,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional JobExecution exec1 = new JobExecution(jobInstance); dao.saveJobExecution(exec1); - JobExecution exec2 = new JobExecution(jobInstance); + JobExecution exec2 = new JobExecution(jobInstance); exec2.setId(exec1.getId()); exec2.incrementVersion(); @@ -266,7 +267,31 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional // expected } - } /* + } + + @Transactional + @Test + public void testSynchronizeStatus() { + + JobExecution exec1 = new JobExecution(jobInstance); + exec1.setStatus(BatchStatus.STARTED); + dao.saveJobExecution(exec1); + + JobExecution exec2 = new JobExecution(jobInstance); + Assert.state(exec1.getId() != null); + exec2.setId(exec1.getId()); + + exec2.setVersion(7); + Assert.state(exec1.getVersion() != exec2.getVersion()); + Assert.state(exec1.getStatus() != exec2.getStatus()); + + dao.synchronizeStatus(exec2); + + assertEquals(exec1.getVersion(), exec2.getVersion()); + assertEquals(exec1.getStatus(), exec2.getStatus()); + } + + /* * Check to make sure the executions are equal. Normally, comparing the id's * is sufficient. However, for testing purposes, especially of a DAO, we * need to make sure all the fields are being stored/retrieved correctly. diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDaoTests.java index 52456c7df..995a3e102 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDaoTests.java @@ -44,6 +44,7 @@ public class JdbcJobExecutionDaoTests extends AbstractJobExecutionDaoTests { dao.saveJobExecution(execution); execution.setStatus(BatchStatus.COMPLETED); + execution.incrementVersion(); dao.synchronizeStatus(execution); assertEquals(BatchStatus.STARTING, execution.getStatus()); }