diff --git a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobExecutionDao.java index 0907c8993..9725148f8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobExecutionDao.java @@ -39,12 +39,12 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements + "where JOB_INSTANCE_ID = ?"; private static final String SAVE_JOB_EXECUTION = "INSERT into %PREFIX%JOB_EXECUTION(JOB_EXECUTION_ID, JOB_INSTANCE_ID, START_TIME, " - + "END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE) values (?, ?, ?, ?, ?, ?, ?, ?)"; + + "END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, VERSION) values (?, ?, ?, ?, ?, ?, ?, ?, ?)"; private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID = ?"; private static final String UPDATE_JOB_EXECUTION = "UPDATE %PREFIX%JOB_EXECUTION set START_TIME = ?, END_TIME = ?, " - + " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ? where JOB_EXECUTION_ID = ?"; + + " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ?, VERSION = ? where JOB_EXECUTION_ID = ?"; private static final String FIND_JOB_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%JOB_EXECUTION" + " where JOB_INSTANCE_ID = ?"; @@ -90,16 +90,18 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements validateJobExecution(jobExecution); + jobExecution.incrementVersion(); + jobExecution.setId(new Long(jobExecutionIncrementer.nextLongValue())); Object[] parameters = new Object[] { jobExecution.getId(), jobExecution.getJobId(), jobExecution.getStartTime(), jobExecution.getEndTime(), jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N", jobExecution.getExitStatus().getExitCode(), - jobExecution.getExitStatus().getExitDescription() }; + jobExecution.getExitStatus().getExitDescription(), jobExecution.getVersion() }; getJdbcTemplate().update( getQuery(SAVE_JOB_EXECUTION), parameters, new int[] { Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, - Types.VARCHAR, Types.VARCHAR }); + Types.VARCHAR, Types.VARCHAR, Types.INTEGER }); } /** @@ -128,6 +130,8 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements validateJobExecution(jobExecution); + jobExecution.incrementVersion(); + String exitDescription = jobExecution.getExitStatus().getExitDescription(); if (exitDescription != null && exitDescription.length() > EXIT_MESSAGE_LENGTH) { exitDescription = exitDescription.substring(0, EXIT_MESSAGE_LENGTH); @@ -135,7 +139,8 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements } Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(), jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N", - jobExecution.getExitStatus().getExitCode(), exitDescription, jobExecution.getId() }; + jobExecution.getExitStatus().getExitCode(), exitDescription, jobExecution.getVersion(), + jobExecution.getId() }; if (jobExecution.getId() == null) { throw new IllegalArgumentException("JobExecution ID cannot be null. JobExecution must be saved " @@ -146,15 +151,14 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements // is invalid and // an exception should be thrown. if (getJdbcTemplate().queryForInt(getQuery(CHECK_JOB_EXECUTION_EXISTS), new Object[] { jobExecution.getId() }) != 1) { - throw new NoSuchObjectException("Invalid JobExecution, ID " + jobExecution.getId() - + " not found."); + throw new NoSuchObjectException("Invalid JobExecution, ID " + jobExecution.getId() + " not found."); } getJdbcTemplate().update( getQuery(UPDATE_JOB_EXECUTION), parameters, new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR, - Types.INTEGER }); + Types.INTEGER, Types.INTEGER }); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobInstanceDao.java b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobInstanceDao.java index da6baafb2..a3f4b21a8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobInstanceDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobInstanceDao.java @@ -32,8 +32,8 @@ import org.springframework.util.Assert; */ public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements JobInstanceDao, InitializingBean { - private static final String CREATE_JOB_INSTANCE = "INSERT into %PREFIX%JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY)" - + " values (?, ?, ?)"; + private static final String CREATE_JOB_INSTANCE = "INSERT into %PREFIX%JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY, VERSION)" + + " values (?, ?, ?, ?)"; private static final String CREATE_JOB_PARAMETERS = "INSERT into %PREFIX%JOB_PARAMS(JOB_INSTANCE_ID, KEY_NAME, TYPE_CD, " + "STRING_VAL, DATE_VAL, LONG_VAL, DOUBLE_VAL) values (?, ?, ?, ?, ?, ?, ?)"; @@ -60,21 +60,23 @@ public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements Assert.state(getJobInstance(job, jobParameters) == null, "JobInstance must not already exist"); Long jobId = new Long(jobIncrementer.nextLongValue()); - Object[] parameters = new Object[] { jobId, job.getName(), createJobKey(jobParameters) }; + + JobInstance jobInstance = new JobInstance(jobId, jobParameters, job); + jobInstance.incrementVersion(); + + Object[] parameters = new Object[] { jobId, job.getName(), createJobKey(jobParameters), jobInstance.getVersion() }; getJdbcTemplate().update(getQuery(CREATE_JOB_INSTANCE), parameters, - new int[] { Types.INTEGER, Types.VARCHAR, Types.VARCHAR }); + new int[] { Types.INTEGER, Types.VARCHAR, Types.VARCHAR, Types.INTEGER }); insertJobParameters(jobId, jobParameters); - JobInstance jobInstance = new JobInstance(jobId, jobParameters, job); return jobInstance; } private String createJobKey(JobParameters jobParameters) { Map props = jobParameters.getParameters(); - // Start with non-empty string for Oracle: - StringBuffer stringBuffer = new StringBuffer("key:"); + StringBuffer stringBuffer = new StringBuffer(); for (Iterator it = props.entrySet().iterator(); it.hasNext();) { Entry entry = (Entry) it.next(); stringBuffer.append(entry.toString() + ";"); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDao.java index d4eb0fce3..0d68d22d2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDao.java @@ -4,66 +4,71 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Set; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.util.Assert; /** * In-memory implementation of {@link JobExecutionDao}. * */ public class MapJobExecutionDao implements JobExecutionDao { - - private static Map executionsByJobInstanceId = TransactionAwareProxyFactory.createTransactionalMap(); + + private static Map executionsById = TransactionAwareProxyFactory.createTransactionalMap(); private static long currentId; public static void clear() { - executionsByJobInstanceId.clear(); + executionsById.clear(); } public int getJobExecutionCount(JobInstance jobInstance) { - Set executions = (Set) executionsByJobInstanceId.get(jobInstance.getId()); - if (executions == null) { - return 0; + int count = 0; + for (Iterator iterator = executionsById.values().iterator(); iterator.hasNext();) { + JobExecution exec = (JobExecution) iterator.next(); + if (exec.getJobInstance().equals(jobInstance)) { + count++; + } } - return executions.size(); + return count; } public void saveJobExecution(JobExecution jobExecution) { - Set executions = (Set) executionsByJobInstanceId.get(jobExecution.getJobId()); - if (executions == null) { - executions = TransactionAwareProxyFactory.createTransactionalSet(); - executionsByJobInstanceId.put(jobExecution.getJobId(), executions); - } - executions.add(jobExecution); - jobExecution.setId(new Long(currentId++)); + Assert.isTrue(jobExecution.getId() == null); + Long newId = new Long(currentId++); + jobExecution.setId(newId); + jobExecution.incrementVersion(); + executionsById.put(newId, jobExecution); } public List findJobExecutions(JobInstance jobInstance) { - Set executions = (Set) executionsByJobInstanceId.get(jobInstance.getId()); - if (executions == null) { - return new ArrayList(); - } - else { - return new ArrayList(executions); + List executions = new ArrayList(); + for (Iterator iterator = executionsById.values().iterator(); iterator.hasNext();) { + JobExecution exec = (JobExecution) iterator.next(); + if (exec.getJobInstance().equals(jobInstance)) { + executions.add(exec); + } } + return executions; } public void updateJobExecution(JobExecution jobExecution) { - // no-op + Long id = jobExecution.getId(); + Assert.notNull(id, "JobExecution is expected to have an id (should be saved already)"); + Assert.notNull(executionsById.get(id), "JobExecution must already be saved"); + jobExecution.incrementVersion(); + executionsById.put(id, jobExecution); } public JobExecution getLastJobExecution(JobInstance jobInstance) { - Set executions = (Set) executionsByJobInstanceId.get(jobInstance.getId()); - if (executions == null) { - return null; - } JobExecution lastExec = null; - for (Iterator iterator = executions.iterator(); iterator.hasNext();) { + for (Iterator iterator = executionsById.values().iterator(); iterator.hasNext();) { JobExecution exec = (JobExecution) iterator.next(); + if (!exec.getJobInstance().equals(jobInstance)) { + continue; + } if (lastExec == null) { lastExec = exec; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDao.java b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDao.java index ab2af2ed4..20b120a2f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDao.java @@ -9,6 +9,9 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.util.Assert; +/** + * In-memory implementation of {@link JobInstanceDao}. + */ public class MapJobInstanceDao implements JobInstanceDao { private static Collection jobInstances = TransactionAwareProxyFactory.createTransactionalList(); @@ -24,6 +27,7 @@ public class MapJobInstanceDao implements JobInstanceDao { Assert.state(getJobInstance(job, jobParameters) == null, "JobInstance must not already exist"); JobInstance jobInstance = new JobInstance(new Long(currentId++), jobParameters, job); + jobInstance.incrementVersion(); jobInstances.add(jobInstance); return jobInstance; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java index a2dace01d..e480c45da 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java @@ -51,8 +51,8 @@ public class MapStepExecutionDao implements StepExecutionDao { } public void saveStepExecution(StepExecution stepExecution) { - Assert.state(stepExecution.getId() == null); - Assert.state(stepExecution.getVersion() == null); + Assert.isTrue(stepExecution.getId() == null); + Assert.isTrue(stepExecution.getVersion() == null); Assert.notNull(stepExecution.getJobExecutionId(), "JobExecution must be saved already."); Map executions = (Map) executionsByJobExecutionId.get(stepExecution.getJobExecutionId()); @@ -60,8 +60,8 @@ public class MapStepExecutionDao implements StepExecutionDao { executions = TransactionAwareProxyFactory.createTransactionalMap(); executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions); } - stepExecution.incrementVersion(); stepExecution.setId(new Long(currentId++)); + stepExecution.incrementVersion(); executions.put(stepExecution.getStepName(), stepExecution); } @@ -75,14 +75,16 @@ public class MapStepExecutionDao implements StepExecutionDao { StepExecution persistedExecution = (StepExecution) executions.get(stepExecution.getStepName()); Assert.notNull(persistedExecution, "step execution is expected to be already saved"); - if (!persistedExecution.getVersion().equals(stepExecution.getVersion())) { - throw new OptimisticLockingFailureException("Attempt to update step execution id=" + stepExecution.getId() - + " with wrong version (" + stepExecution.getVersion() + "), where current version is " - + persistedExecution.getVersion()); - } + synchronized (stepExecution) { + if (!persistedExecution.getVersion().equals(stepExecution.getVersion())) { + throw new OptimisticLockingFailureException("Attempt to update step execution id=" + + stepExecution.getId() + " with wrong version (" + stepExecution.getVersion() + + "), where current version is " + persistedExecution.getVersion()); + } - stepExecution.incrementVersion(); - executions.put(stepExecution.getStepName(), stepExecution); + stepExecution.incrementVersion(); + executions.put(stepExecution.getStepName(), stepExecution); + } } public StepExecution getStepExecution(JobExecution jobExecution, Step step) { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/AbstractJobDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/AbstractJobDaoTests.java index 99f325013..f2db753e3 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/AbstractJobDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/AbstractJobDaoTests.java @@ -284,4 +284,32 @@ public abstract class AbstractJobDaoTests extends AbstractTransactionalDataSourc // expected } } + + public void testCreationAddsVersion() { + + jobInstance = jobInstanceDao.createJobInstance(new JobSupport("testVersion"), new JobParameters()); + + assertNotNull(jobInstance.getVersion()); + } + + public void testSaveAddsVersionAndId() { + + JobExecution jobExecution = new JobExecution(jobInstance); + + assertNull(jobExecution.getId()); + assertNull(jobExecution.getVersion()); + + jobExecutionDao.saveJobExecution(jobExecution); + + assertNotNull(jobExecution.getId()); + assertNotNull(jobExecution.getVersion()); + } + + public void testUpdateIncrementsVersion() { + int version = jobExecution.getVersion().intValue(); + + jobExecutionDao.updateJobExecution(jobExecution); + + assertEquals(version + 1, jobExecution.getVersion().intValue()); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDaoTests.java index edd868351..f86debe7d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDaoTests.java @@ -38,11 +38,13 @@ public class MapJobExecutionDaoTests extends TestCase { /** * Saving sets id to the entity. */ - public void testSaveAddsId() { + public void testSaveAddsIdAndVersion() { assertNull(execution.getId()); + assertNull(execution.getVersion()); dao.saveJobExecution(execution); assertNotNull(execution.getId()); + assertNotNull(execution.getVersion()); } /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDaoTests.java index b9f9a96e2..0d3bdb5f7 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDaoTests.java @@ -42,9 +42,9 @@ public class MapJobInstanceDaoTests extends TestCase { * Trying to create instance twice for the same job+parameters causes error */ public void testCreateDuplicateInstance() { - + dao.createJobInstance(fooJob, fooParams); - + try { dao.createJobInstance(fooJob, fooParams); fail(); @@ -53,4 +53,15 @@ public class MapJobInstanceDaoTests extends TestCase { // expected } } + + public void testCreationAddsVersion() { + + JobInstance jobInstance = new JobInstance(new Long(1), new JobParameters(), new JobSupport("testVersionAndId")); + + assertNull(jobInstance.getVersion()); + + jobInstance = dao.createJobInstance(new JobSupport("testVersion"), new JobParameters()); + + assertNotNull(jobInstance.getVersion()); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java index 44658974e..0005fad49 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDaoTests.java @@ -31,6 +31,9 @@ import org.springframework.batch.execution.step.StepSupport; import org.springframework.batch.item.ExecutionContext; import org.springframework.dao.OptimisticLockingFailureException; +/** + * Tests for {@link MapStepExecutionDao}. + */ public class MapStepExecutionDaoTests extends TestCase { private StepExecutionDao dao = new MapStepExecutionDao(); @@ -51,11 +54,12 @@ public class MapStepExecutionDaoTests extends TestCase { stepExecution = new StepExecution(step, jobExecution); } - public void testSaveExecutionUpdatesId() throws Exception { - StepExecution execution = new StepExecution(step, new JobExecution(jobInstance, new Long(1))); - assertNull(execution.getId()); - dao.saveStepExecution(execution); - assertNotNull(execution.getId()); + public void testSaveExecutionAssignsIdAndVersion() throws Exception { + assertNull(stepExecution.getId()); + assertNull(stepExecution.getVersion()); + dao.saveStepExecution(stepExecution); + assertNotNull(stepExecution.getId()); + assertNotNull(stepExecution.getVersion()); } public void testSaveAndFindExecution() { @@ -65,14 +69,54 @@ public class MapStepExecutionDaoTests extends TestCase { StepExecution retrieved = dao.getStepExecution(jobExecution, step); assertEquals(stepExecution, retrieved); assertEquals(BatchStatus.STARTED, retrieved.getStatus()); + + assertNull(dao.getStepExecution(jobExecution, new StepSupport("not-existing step"))); } + public void testGetForNotExistingJobExecution() { + assertNull(dao.getStepExecution(new JobExecution(jobInstance, new Long(777)), step)); + } + + /** + * To-be-saved execution must not already have an id. + */ + public void testSaveExecutionWithIdAlreadySet() { + stepExecution.setId(new Long(7)); + try { + dao.saveStepExecution(stepExecution); + fail(); + } + catch (IllegalArgumentException e) { + // expected + } + } + + /** + * To-be-saved execution must not already have a version. + */ + public void testSaveExecutionWithVersionAlreadySet() { + stepExecution.incrementVersion(); + try { + dao.saveStepExecution(stepExecution); + fail(); + } + catch (IllegalArgumentException e) { + // expected + } + } + + /** + * Update and retrieve updated StepExecution - make sure the update is + * reflected as expected and version number has been incremented + */ public void testUpdateExecution() { stepExecution.setStatus(BatchStatus.STARTED); dao.saveStepExecution(stepExecution); + Integer versionAfterSave = stepExecution.getVersion(); stepExecution.setStatus(BatchStatus.STOPPED); dao.updateStepExecution(stepExecution); + assertEquals(versionAfterSave.intValue() + 1, stepExecution.getVersion().intValue()); StepExecution retrieved = dao.getStepExecution(jobExecution, step); assertEquals(stepExecution, retrieved); @@ -109,6 +153,10 @@ public class MapStepExecutionDaoTests extends TestCase { assertEquals(7, retrieved.getLong("longKey")); } + /** + * Exception should be raised when the version of update argument doesn't + * match the version of persisted entity. + */ public void testConcurrentModificationException() { jobInstance = new JobInstance(new Long(1), new JobParameters(), new JobSupport("testJob")); jobExecution = new JobExecution(jobInstance, new Long(1));