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 3b0e29337..03ce69068 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 @@ -15,6 +15,7 @@ import org.springframework.batch.core.JobInstance; import org.springframework.batch.repeat.ExitStatus; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.EmptyResultDataAccessException; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.jdbc.core.RowCallbackHandler; import org.springframework.jdbc.core.simple.ParameterizedRowMapper; import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; @@ -41,23 +42,25 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements + "END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, VERSION, CREATE_TIME, LAST_UPDATED) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID = ?"; - + private static final String GET_STATUS = "SELECT STATUS 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 = ?, VERSION = ?, CREATE_TIME = ?, LAST_UPDATED = ? where JOB_EXECUTION_ID = ?"; + + " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ?, VERSION = ?, CREATE_TIME = ?, LAST_UPDATED = ? where JOB_EXECUTION_ID = ? and VERSION = ?"; - private static final String FIND_JOB_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED" + - " from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? order by JOB_EXECUTION_ID desc"; + private static final String FIND_JOB_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION" + + " from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? order by JOB_EXECUTION_ID desc"; - private static final String GET_LAST_EXECUTION = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED " + - "from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? and CREATE_TIME = (SELECT max(CREATE_TIME) from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ?)"; + private static final String GET_LAST_EXECUTION = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION " + + "from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? and CREATE_TIME = (SELECT max(CREATE_TIME) from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ?)"; - private static final String GET_EXECUTION_BY_ID = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED" + - " from %PREFIX%JOB_EXECUTION where JOB_EXECUTION_ID = ?"; + private static final String GET_EXECUTION_BY_ID = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION" + + " from %PREFIX%JOB_EXECUTION where JOB_EXECUTION_ID = ?"; - private static final String GET_RUNNING_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, " + - "JOB_INSTANCE_ID from %PREFIX%JOB_EXECUTION where END_TIME is NULL order by JOB_EXECUTION_ID desc"; + private static final String GET_RUNNING_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION " + + "JOB_INSTANCE_ID from %PREFIX%JOB_EXECUTION where END_TIME is NULL order by JOB_EXECUTION_ID desc"; + + private static final String CURRENT_VERSION_JOB_EXECUTION = "SELECT VERSION FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID=?"; private int exitMessageLength = DEFAULT_EXIT_MESSAGE_LENGTH; @@ -151,35 +154,52 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements validateJobExecution(jobExecution); - jobExecution.incrementVersion(); + Assert.notNull(jobExecution.getId(), + "JobExecution ID cannot be null. JobExecution must be saved before it can be updated"); - String exitDescription = jobExecution.getExitStatus().getExitDescription(); - if (exitDescription != null && exitDescription.length() > exitMessageLength) { - exitDescription = exitDescription.substring(0, exitMessageLength); - logger.debug("Truncating long message before update of JobExecution: " + jobExecution); + Assert.notNull(jobExecution.getVersion(), + "JobExecution version cannot be null. JobExecution must be saved before it can be updated"); + + synchronized (jobExecution) { + Integer version = jobExecution.getVersion() + 1; + + String exitDescription = jobExecution.getExitStatus().getExitDescription(); + if (exitDescription != null && exitDescription.length() > exitMessageLength) { + exitDescription = exitDescription.substring(0, exitMessageLength); + logger.debug("Truncating long message before update of JobExecution: " + jobExecution); + } + Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(), + jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N", + jobExecution.getExitStatus().getExitCode(), exitDescription, version, jobExecution.getCreateTime(), + jobExecution.getLastUpdated(), jobExecution.getId(), jobExecution.getVersion() }; + + // Check if given JobExecution's Id already exists, if none is found + // it + // 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."); + } + + int count = getJdbcTemplate().getJdbcOperations().update( + getQuery(UPDATE_JOB_EXECUTION), + parameters, + new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, Types.VARCHAR, + Types.VARCHAR, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.INTEGER, + Types.INTEGER }); + + // Avoid concurrent modifications... + if (count == 0) { + int curentVersion = getJdbcTemplate().queryForInt(getQuery(CURRENT_VERSION_JOB_EXECUTION), + new Object[] { jobExecution.getId() }); + throw new OptimisticLockingFailureException("Attempt to update step execution id=" + + jobExecution.getId() + " with wrong version (" + jobExecution.getVersion() + + "), where current version is " + curentVersion); + } + + jobExecution.incrementVersion(); } - Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(), - jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N", - jobExecution.getExitStatus().getExitCode(), exitDescription, jobExecution.getVersion(), - jobExecution.getCreateTime(), jobExecution.getLastUpdated(), jobExecution.getId() }; - - if (jobExecution.getId() == null) { - throw new IllegalArgumentException("JobExecution ID cannot be null. JobExecution must be saved " - + "before it can be updated."); - } - - // Check if given JobExecution's Id already exists, if none is found it - // 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."); - } - - getJdbcTemplate().getJdbcOperations().update( - getQuery(UPDATE_JOB_EXECUTION), - parameters, - new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR, - Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.INTEGER }); } public JobExecution getLastJobExecution(JobInstance jobInstance) { @@ -235,9 +255,9 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements return result; } - + public void synchronizeStatus(JobExecution jobExecution) { - + String status = getJdbcTemplate().queryForObject(getQuery(GET_STATUS), String.class, jobExecution.getId()); jobExecution.setStatus(BatchStatus.valueOf(status)); } @@ -251,31 +271,32 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements private class JobExecutionRowMapper implements ParameterizedRowMapper { private JobInstance jobInstance; - + public JobExecutionRowMapper() { } - + public JobExecutionRowMapper(JobInstance jobInstance) { this.jobInstance = jobInstance; } - + public JobExecution mapRow(ResultSet rs, int rowNum) throws SQLException { Long id = new Long(rs.getLong(1)); - JobExecution jobExecution; - - if(jobInstance == null){ + JobExecution jobExecution; + + if (jobInstance == null) { jobExecution = new JobExecution(id); } - else{ + else { jobExecution = new JobExecution(jobInstance, id); } - + jobExecution.setStartTime(rs.getTimestamp(2)); jobExecution.setEndTime(rs.getTimestamp(3)); jobExecution.setStatus(BatchStatus.valueOf(rs.getString(4))); jobExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(5)), rs.getString(6), rs.getString(7))); jobExecution.setCreateTime(rs.getTimestamp(8)); jobExecution.setLastUpdated(rs.getTimestamp(9)); + jobExecution.setVersion(rs.getInt(10)); return jobExecution; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java index bdd8890e6..d02b859f8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java @@ -130,12 +130,6 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null."); } - /* - * (non-Javadoc) - * - * @seeorg.springframework.batch.execution.repository.dao.StepExecutionDao# - * updateStepExecution(org.springframework.batch.core.domain.StepExecution) - */ public void updateStepExecution(StepExecution stepExecution) { validateStepExecution(stepExecution); @@ -151,7 +145,7 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement // someone is already trying to do it. synchronized (stepExecution) { - Integer version = new Integer(stepExecution.getVersion().intValue() + 1); + Integer version = stepExecution.getVersion() + 1; Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getReadCount(), stepExecution.getFilterCount(), stepExecution.getWriteCount(), 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 ae1464fc4..31f87518b 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 @@ -12,6 +12,7 @@ import org.apache.commons.lang.SerializationUtils; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.util.Assert; /** @@ -68,9 +69,18 @@ public class MapJobExecutionDao implements JobExecutionDao { public void updateJobExecution(JobExecution jobExecution) { 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, copy(jobExecution)); + JobExecution persistedExecution = executionsById.get(id); + Assert.notNull(persistedExecution, "JobExecution must already be saved"); + + synchronized (jobExecution) { + 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()); + } + jobExecution.incrementVersion(); + executionsById.put(id, copy(jobExecution)); + } } public JobExecution getLastJobExecution(JobInstance jobInstance) { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobDaoTests.java index 6c07b849a..02f6808cb 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobDaoTests.java @@ -190,6 +190,7 @@ public abstract class AbstractJobDaoTests { // id is invalid JobExecution execution = new JobExecution(jobInstance, (long) 29432); + execution.incrementVersion(); try { jobExecutionDao.updateJobExecution(execution); fail("Expected NoSuchBatchDomainObjectException"); 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 9549844cc..852fe9918 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 @@ -1,9 +1,9 @@ package org.springframework.batch.core.repository.dao; -import static org.junit.Assert.*; - -import org.junit.Before; -import org.junit.Test; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.Collections; @@ -11,22 +11,25 @@ import java.util.Date; import java.util.List; import java.util.Set; +import org.junit.Before; +import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.repeat.ExitStatus; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.test.context.junit4.AbstractTransactionalJUnit4SpringContextTests; import org.springframework.transaction.annotation.Transactional; public abstract class AbstractJobExecutionDaoTests extends AbstractTransactionalJUnit4SpringContextTests { - JobExecutionDao dao; + protected JobExecutionDao dao; - JobInstance jobInstance = new JobInstance((long) 1, new JobParameters(), "execTestJob"); + protected JobInstance jobInstance = new JobInstance((long) 1, new JobParameters(), "execTestJob"); - JobExecution execution = new JobExecution(jobInstance); + protected JobExecution execution = new JobExecution(jobInstance); /** * @return tested object ready for use @@ -63,31 +66,30 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional assertEquals(execution, executions.get(0)); assertExecutionsAreEqual(execution, executions.get(0)); } - + /** * Executions should be returned in the reverse order they were saved. */ @Transactional @Test public void testFindExecutionsOrdering() { - + List execs = new ArrayList(); - + for (int i = 0; i < 10; i++) { JobExecution exec = new JobExecution(jobInstance); exec.setCreateTime(new Date(i)); execs.add(exec); dao.saveJobExecution(exec); } - + List retrieved = dao.findJobExecutions(jobInstance); Collections.reverse(retrieved); - - + for (int i = 0; i < 10; i++) { assertExecutionsAreEqual(execs.get(i), retrieved.get(i)); } - + } /** @@ -162,7 +164,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional JobExecution value = dao.getLastJobExecution(jobInstance); assertNull(value); } - + /** * Check the execution is returned */ @@ -189,7 +191,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional assertEquals(1, values.size()); JobExecution value = values.iterator().next(); assertEquals(exec, value); - assertEquals(5L, value.getLastUpdated().getTime()); + assertEquals(5L, value.getLastUpdated().getTime()); } /** @@ -201,7 +203,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional Set values = dao.findRunningJobExecutions("no-such-job"); assertEquals(0, values.size()); } - + /** * Check the execution is returned */ @@ -233,20 +235,51 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional JobExecution value = dao.getJobExecution(54321L); assertNull(value); } - - /* - * 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. + + /** + * Exception should be raised when the version of update argument doesn't + * match the version of persisted entity. */ - private void assertExecutionsAreEqual(JobExecution lhs, JobExecution rhs){ - + @Transactional + @Test + public void testConcurrentModificationException() { + + JobExecution exec1 = new JobExecution(jobInstance); + dao.saveJobExecution(exec1); + + JobExecution exec2 = new JobExecution(jobInstance); + exec2.setId(exec1.getId()); + + exec2.incrementVersion(); + assertEquals((Integer) 0, exec1.getVersion()); + assertEquals(exec1.getVersion(), exec2.getVersion()); + + dao.updateJobExecution(exec1); + assertEquals((Integer) 1, exec1.getVersion()); + + try { + dao.updateJobExecution(exec2); + fail(); + } + catch (OptimisticLockingFailureException e) { + // expected + } + + } /* + * 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. + */ + + private void assertExecutionsAreEqual(JobExecution lhs, JobExecution rhs) { + assertEquals(lhs.getId(), rhs.getId()); assertEquals(lhs.getStartTime(), rhs.getStartTime()); assertEquals(lhs.getStatus(), rhs.getStatus()); assertEquals(lhs.getEndTime(), rhs.getEndTime()); assertEquals(lhs.getCreateTime(), rhs.getCreateTime()); assertEquals(lhs.getLastUpdated(), rhs.getLastUpdated()); + assertEquals(lhs.getVersion(), rhs.getVersion()); } - + } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/DatabaseShutdownFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/DatabaseShutdownFunctionalTests.java index 1012668d9..0fd2e438c 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/DatabaseShutdownFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/DatabaseShutdownFunctionalTests.java @@ -16,9 +16,7 @@ package org.springframework.batch.sample; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.*; import org.junit.Test; import org.junit.runner.RunWith; @@ -59,6 +57,7 @@ public class DatabaseShutdownFunctionalTests extends AbstractBatchLauncherTests assertEquals(BatchStatus.STARTED, jobExecution.getStatus()); assertTrue(jobExecution.isRunning()); + assertNotNull(jobExecution.getVersion()); jobOperator.stop(jobExecution.getId());