From 79e7ebedd406767445faa2a2074c744268e028b1 Mon Sep 17 00:00:00 2001 From: dsyer Date: Wed, 19 Dec 2007 21:01:32 +0000 Subject: [PATCH] Re-instated changes for version check and optimistic locking --- .../execution/repository/dao/JdbcStepDao.java | 187 ++++++++---------- .../repository/dao/JdbcJobDaoQueryTests.java | 13 +- 2 files changed, 85 insertions(+), 115 deletions(-) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java index 1606de65b..44e2e9c6c 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java @@ -36,6 +36,7 @@ import org.springframework.batch.restart.RestartData; import org.springframework.batch.support.PropertiesConverter; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.IncorrectResultSizeDataAccessException; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; @@ -46,11 +47,11 @@ import org.springframework.util.StringUtils; * Sql implementation of {@link StepDao}. Uses Sequences (via Spring's * * @link DataFieldMaxValueIncrementer abstraction) to create all Step and - * StepExecution primary keys before inserting a new row. All objects are - * checked to ensure all fields to be stored are not null. If any are - * found to be null, an IllegalArgumentException will be thrown. This - * could be left to JdbcTemplate, however, the exception will be fairly - * vague, and fails to highlight which field caused the exception. + * StepExecution primary keys before inserting a new row. All objects are + * checked to ensure all fields to be stored are not null. If any are found to + * be null, an IllegalArgumentException will be thrown. This could be left to + * JdbcTemplate, however, the exception will be fairly vague, and fails to + * highlight which field caused the exception. * * TODO: JavaDoc should be geared more towards usability, the comments above are * useful information, and should be there, but needs usability stuff. Depends @@ -89,7 +90,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, " + "STATUS = ?, COMMIT_COUNT = ?, TASK_COUNT = ?, TASK_STATISTICS = ?, CONTINUABLE = ? , EXIT_CODE = ?, " - + "EXIT_MESSAGE = ? where ID = ?"; + + "EXIT_MESSAGE = ?, VERSION = ? where ID = ? and VERSION = ?"; private JdbcOperations jdbcTemplate; @@ -104,8 +105,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { public void afterPropertiesSet() throws Exception { Assert.notNull(jdbcTemplate, "JdbcTemplate cannot be null."); Assert.notNull(stepIncrementer, "StepIncrementer cannot be null."); - Assert.notNull(stepExecutionIncrementer, - "StepExecutionIncrementer canot be null."); + Assert.notNull(stepExecutionIncrementer, "StepExecutionIncrementer canot be null."); } private void cascadeJobExecution(JobExecution jobExecution) { @@ -122,8 +122,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { * DataFieldMaxValueIncrementer) * * @see StepDao#createStep(JobInstance, String) - * @throws IllegalArgumentException - * if job or stepName is null. + * @throws IllegalArgumentException if job or stepName is null. */ public StepInstance createStep(JobInstance job, String stepName) { @@ -145,10 +144,9 @@ public class JdbcStepDao implements StepDao, InitializingBean { * anymore than one step is found, an exception is thrown. * * @see StepDao#findStep(Long, String) - * @throws IllegalArgumentException - * if job, stepName, or job.id is null. - * @throws IncorrectResultSizeDataAccessException - * if more than one step is found. + * @throws IllegalArgumentException if job, stepName, or job.id is null. + * @throws IncorrectResultSizeDataAccessException if more than one step is + * found. */ public StepInstance findStep(JobInstance job, String stepName) { @@ -164,30 +162,28 @@ public class JdbcStepDao implements StepDao, InitializingBean { StepInstance step = new StepInstance(new Long(rs.getLong(1))); step.setStatus(BatchStatus.getStatus(rs.getString(2))); - step.setRestartData(new GenericRestartData(PropertiesConverter - .stringToProperties(rs.getString(3)))); + step.setRestartData(new GenericRestartData(PropertiesConverter.stringToProperties(rs.getString(3)))); return step; } }; - List steps = jdbcTemplate.query(getFindStepQuery(), parameters, - rowMapper); + List steps = jdbcTemplate.query(getFindStepQuery(), parameters, rowMapper); if (steps.size() == 0) { // No step found return null; - } else if (steps.size() == 1) { + } + else if (steps.size() == 1) { StepInstance step = (StepInstance) steps.get(0); return step; - } else { + } + else { // This error will likely never be thrown, because there should // never be two steps with the same name and Job_ID due to database // constraints. - throw new IncorrectResultSizeDataAccessException( - "Step Invalid, multiple steps found for StepName:" - + stepName + " and JobId:" + job.getId(), 1, steps - .size()); + throw new IncorrectResultSizeDataAccessException("Step Invalid, multiple steps found for StepName:" + + stepName + " and JobId:" + job.getId(), 1, steps.size()); } } @@ -197,8 +193,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { * they will not be returned with reconstituted object. * * @see StepDao#getStepExecution(Long) - * @throws IllegalArgumentException - * if id is null. + * @throws IllegalArgumentException if id is null. */ public List findStepExecutions(final StepInstance step) { @@ -208,28 +203,23 @@ public class JdbcStepDao implements StepDao, InitializingBean { RowMapper rowMapper = new RowMapper() { public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - JobExecution jobExecution = (JobExecution) jdbcTemplate - .queryForObject( - getQuery(JobExecutionRowMapper.GET_JOB_EXECUTION), - new Object[] { new Long(rs.getLong(2)) }, - new JobExecutionRowMapper(step.getJob())); - StepExecution stepExecution = new StepExecution(step, - jobExecution, new Long(rs.getLong(1))); + JobExecution jobExecution = (JobExecution) jdbcTemplate.queryForObject( + getQuery(JobExecutionRowMapper.GET_JOB_EXECUTION), new Object[] { new Long(rs.getLong(2)) }, + new JobExecutionRowMapper(step.getJob())); + StepExecution stepExecution = new StepExecution(step, jobExecution, new Long(rs.getLong(1))); stepExecution.setStartTime(rs.getTimestamp(3)); stepExecution.setEndTime(rs.getTimestamp(4)); stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5))); stepExecution.setCommitCount(rs.getInt(6)); stepExecution.setTaskCount(rs.getInt(7)); - stepExecution.setStatistics(PropertiesConverter - .stringToProperties(rs.getString(8))); - stepExecution.setExitStatus(new ExitStatus("Y".equals(rs - .getString(9)), rs.getString(10), rs.getString(11))); + stepExecution.setStatistics(PropertiesConverter.stringToProperties(rs.getString(8))); + stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(9)), rs.getString(10), rs + .getString(11))); return stepExecution; } }; - return jdbcTemplate.query(getFindStepExecutionsQuery(), - new Object[] { step.getId() }, rowMapper); + return jdbcTemplate.query(getFindStepExecutionsQuery(), new Object[] { step.getId() }, rowMapper); } @@ -239,8 +229,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { * Sql implementation which uses a RowMapper to populate a list of all rows * in the step table with the same JOB_ID. * - * @throws IllegalArgumentException - * if jobId is null. + * @throws IllegalArgumentException if jobId is null. */ public List findSteps(final JobInstance job) { @@ -252,12 +241,10 @@ public class JdbcStepDao implements StepDao, InitializingBean { public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - StepInstance step = new StepInstance(job, rs.getString(2), - new Long(rs.getLong(1))); + StepInstance step = new StepInstance(job, rs.getString(2), new Long(rs.getLong(1))); String status = rs.getString(3); step.setStatus(BatchStatus.getStatus(status)); - step.setRestartData(new GenericRestartData(PropertiesConverter - .stringToProperties(rs.getString(3)))); + step.setRestartData(new GenericRestartData(PropertiesConverter.stringToProperties(rs.getString(3)))); return step; } }; @@ -293,8 +280,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { Object[] parameters = new Object[] { stepId }; - return jdbcTemplate.queryForInt(getStepExecutionCountQuery(), - parameters); + return jdbcTemplate.queryForInt(getStepExecutionCountQuery(), parameters); } private String getStepExecutionCountQuery() { @@ -323,26 +309,16 @@ public class JdbcStepDao implements StepDao, InitializingBean { cascadeJobExecution(stepExecution.getJobExecution()); stepExecution.setId(new Long(stepExecutionIncrementer.nextLongValue())); - Object[] parameters = new Object[] { - stepExecution.getId(), - new Long(0), - stepExecution.getStepId(), - stepExecution.getJobExecutionId(), - stepExecution.getStartTime(), - stepExecution.getEndTime(), - stepExecution.getStatus().toString(), - stepExecution.getCommitCount(), - stepExecution.getTaskCount(), - PropertiesConverter.propertiesToString(stepExecution - .getStatistics()), - stepExecution.getExitStatus().isContinuable() ? "Y" : "N", - stepExecution.getExitStatus().getExitCode(), + stepExecution.incrementVersion(); // should be 0 now + Object[] parameters = new Object[] { stepExecution.getId(), stepExecution.getVersion(), stepExecution.getStepId(), + stepExecution.getJobExecutionId(), stepExecution.getStartTime(), stepExecution.getEndTime(), + stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getTaskCount(), + PropertiesConverter.propertiesToString(stepExecution.getStatistics()), + stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(), stepExecution.getExitStatus().getExitDescription() }; - jdbcTemplate.update(getSaveStepExecutionQuery(), parameters, new int[] { - Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.INTEGER, - Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, - Types.INTEGER, Types.VARCHAR, Types.CHAR, Types.VARCHAR, - Types.VARCHAR }); + jdbcTemplate.update(getSaveStepExecutionQuery(), parameters, new int[] { Types.INTEGER, Types.INTEGER, + Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, + Types.INTEGER, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR }); } @@ -354,8 +330,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { * Injection setter for job dao. Used to save {@link JobExecution} * instances. * - * @param jobDao - * a {@link JobDao} + * @param jobDao a {@link JobDao} */ public void setJobDao(JobDao jobDao) { this.jobDao = jobDao; @@ -367,8 +342,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { * * @param stepExecutionIncrementer a {@link DataFieldMaxValueIncrementer} */ - public void setStepExecutionIncrementer( - DataFieldMaxValueIncrementer stepExecutionIncrementer) { + public void setStepExecutionIncrementer(DataFieldMaxValueIncrementer stepExecutionIncrementer) { this.stepExecutionIncrementer = stepExecutionIncrementer; } @@ -388,8 +362,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { * are overridden with the set*Query methods). Defaults to * {@value #DEFAULT_TABLE_PREFIX}. * - * @param tablePrefix - * the tablePrefix to set + * @param tablePrefix the tablePrefix to set */ public void setTablePrefix(String tablePrefix) { this.tablePrefix = tablePrefix; @@ -401,9 +374,8 @@ public class JdbcStepDao implements StepDao, InitializingBean { public void update(StepExecution stepExecution) { validateStepExecution(stepExecution); - Assert.notNull(stepExecution.getId(), - "StepExecution Id cannot be null. StepExecution must saved" - + " before it can be updated."); + Assert.notNull(stepExecution.getId(), "StepExecution Id cannot be null. StepExecution must saved" + + " before it can be updated."); // TODO: Not sure if this is a good idea on step execution considering // it is saved at every commit @@ -413,40 +385,41 @@ public class JdbcStepDao implements StepDao, InitializingBean { // return; // throw exception? // } - String exitDescription = stepExecution.getExitStatus() - .getExitDescription(); - if (exitDescription != null - && exitDescription.length() > EXIT_MESSAGE_LENGTH) { + String exitDescription = stepExecution.getExitStatus().getExitDescription(); + if (exitDescription != null && exitDescription.length() > EXIT_MESSAGE_LENGTH) { exitDescription = exitDescription.substring(0, EXIT_MESSAGE_LENGTH); - logger - .debug("Truncating long message before update of StepExecution: " - + stepExecution); + logger.debug("Truncating long message before update of StepExecution: " + stepExecution); } - Object[] parameters = new Object[] { - stepExecution.getStartTime(), - stepExecution.getEndTime(), - stepExecution.getStatus().toString(), - stepExecution.getCommitCount(), - stepExecution.getTaskCount(), - PropertiesConverter.propertiesToString(stepExecution - .getStatistics()), - stepExecution.getExitStatus().isContinuable() ? "Y" : "N", - stepExecution.getExitStatus().getExitCode(), exitDescription, - stepExecution.getId() }; - jdbcTemplate - .update(getUpdateStepExecutionQuery(), parameters, - new int[] { Types.TIMESTAMP, Types.TIMESTAMP, - Types.VARCHAR, Types.INTEGER, Types.INTEGER, - Types.VARCHAR, Types.CHAR, Types.VARCHAR, - Types.VARCHAR, Types.INTEGER }); + // Attempt to prevent concurrent modification errors by blocking here if + // someone is already trying to do it. + synchronized (stepExecution) { + Integer version = new Integer(stepExecution.getVersion().intValue() + 1); + Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(), + stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getTaskCount(), + PropertiesConverter.propertiesToString(stepExecution.getStatistics()), + stepExecution.getExitStatus().isContinuable() ? "Y" : "N", + stepExecution.getExitStatus().getExitCode(), exitDescription, version, stepExecution.getId(), + stepExecution.getVersion() }; + int count = jdbcTemplate.update(getUpdateStepExecutionQuery(), parameters, new int[] { Types.TIMESTAMP, + Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.CHAR, + Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER }); + + // Avoid concurrent modifications... + if (count == 0) { + throw new OptimisticLockingFailureException("Attempt to update step execution id=" + + stepExecution.getId() + " with out of date version (" + stepExecution.getVersion() + ")"); + } + + stepExecution.incrementVersion(); + + } } /** * @see StepDao#update(StepInstance) - * @throws IllegalArgumentException - * if step, or it's status and id is null. + * @throws IllegalArgumentException if step, or it's status and id is null. */ public void update(final StepInstance step) { @@ -461,8 +434,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { } Object[] parameters = new Object[] { step.getStatus().toString(), - PropertiesConverter.propertiesToString(restartProps), - step.getId() }; + PropertiesConverter.propertiesToString(restartProps), step.getId() }; jdbcTemplate.update(getUpdateStepQuery(), parameters); } @@ -476,12 +448,9 @@ public class JdbcStepDao implements StepDao, InitializingBean { private void validateStepExecution(StepExecution stepExecution) { Assert.notNull(stepExecution); - Assert.notNull(stepExecution.getStepId(), - "StepExecution Step-Id cannot be null."); - Assert.notNull(stepExecution.getStartTime(), - "StepExecution start time cannot be null."); - Assert.notNull(stepExecution.getStatus(), - "StepExecution status cannot be null."); + Assert.notNull(stepExecution.getStepId(), "StepExecution Step-Id cannot be null."); + Assert.notNull(stepExecution.getStartTime(), "StepExecution start time cannot be null."); + Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null."); } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcJobDaoQueryTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcJobDaoQueryTests.java index 712d1472a..3375aaf3b 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcJobDaoQueryTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcJobDaoQueryTests.java @@ -28,14 +28,16 @@ import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer /** * @author Dave Syer - * + * */ public class JdbcJobDaoQueryTests extends TestCase { JdbcJobDao sqlDao; + List list = new ArrayList(); - /* (non-Javadoc) + /* + * (non-Javadoc) * @see junit.framework.TestCase#setUp() */ protected void setUp() throws Exception { @@ -53,15 +55,14 @@ public class JdbcJobDaoQueryTests extends TestCase { public String nextStringValue() throws DataAccessException { return "bar"; } - + }); } public void testTablePrefix() throws Exception { sqlDao.setTablePrefix("FOO_"); sqlDao.setJdbcTemplate(new JdbcTemplate() { - public int update(String sql, Object[] args, int[] argTypes) - throws DataAccessException { + public int update(String sql, Object[] args, int[] argTypes) throws DataAccessException { list.add(sql); return 1; } @@ -69,7 +70,7 @@ public class JdbcJobDaoQueryTests extends TestCase { sqlDao.save(new JobInstance(new SimpleJobIdentifier("foo"), new Long(11)).createJobExecution()); assertEquals(1, list.size()); String query = (String) list.get(0); - assertTrue("Query did not contain FOO_:"+query, query.indexOf("FOO_")>=0); + assertTrue("Query did not contain FOO_:" + query, query.indexOf("FOO_") >= 0); } }