diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Entity.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Entity.java index cd9cbfadf..ef8222f45 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Entity.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Entity.java @@ -34,7 +34,7 @@ public class Entity implements Serializable { private Long id; - private Integer version; + private Integer version = new Integer(0); public Entity() { super(); @@ -59,6 +59,13 @@ public class Entity implements Serializable { public Integer getVersion() { return version; } + + /** + * + */ + public void incrementVersion() { + version = new Integer(version.intValue()+1); + } // @Override public String toString() { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobInstance.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobInstance.java index 8e8435592..11ec7f987 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobInstance.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobInstance.java @@ -101,7 +101,7 @@ public class JobInstance extends Entity { return identifier==null ? null : identifier.getName(); } - public JobExecution createNewJobExecution() { + public JobExecution createJobExecution() { return new JobExecution(this); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepContribution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepContribution.java new file mode 100644 index 000000000..6dcd9388b --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepContribution.java @@ -0,0 +1,128 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.core.domain; + +import java.util.Properties; + +import org.springframework.batch.repeat.RepeatContext; + +/** + * Represents a contribution to a {@link StepExecution}, buffering changes + * until they can be applied at a chunk boundary. + * + * @author Dave Syer + * + */ +public class StepContribution { + + /** + * Context attribute key for step execution. Used by monitoring and managing + * clients to inspect current step execution. + */ + private static final String STEP_EXECUTION_KEY = "STEP_EXECUTION"; + + private int taskCount = 0; + + private StepExecution execution; + + private Properties statistics; + + private int commitCount; + + /** + * @param execution + */ + public StepContribution(StepExecution execution) { + this.execution = execution; + } + + /** + * Increment the counter for the number of tasks executed. + */ + public void incrementTaskCount() { + taskCount++; + } + + /** + * Public access to the task execution counter. + * + * @return the task execution counter. + */ + public int getTaskCount() { + return taskCount; + } + + /** + * @param context + */ + public void registerChunkContext(final RepeatContext context) { + execution.getJobExecution().registerChunkContext(context); + context.registerDestructionCallback("CHUNK_EXECUTION_CONTEXT_CALLBACK", new Runnable() { + public void run() { + execution.getJobExecution().unregisterStepContext(context); + } + }); + + } + + /** + * @param context + */ + public void registerStepContext(final RepeatContext context) { + execution.getJobExecution().registerStepContext(context); + context.registerDestructionCallback("STEP_EXECUTION_CONTEXT_CALLBACK", new Runnable() { + public void run() { + execution.getJobExecution().unregisterStepContext(context); + } + }); + // Add the step execution as an attribute so monitoring + // clients can see it. + context.setAttribute(STEP_EXECUTION_KEY, execution); + } + + /** + * Set the statistics properties. + * + * @param statistics + */ + public void setStatistics(Properties statistics) { + this.statistics = statistics; + } + + /** + * Increment the commit counter. + */ + public void incrementCommitCount() { + commitCount++; + } + + /** + * Public getter for the statistics. + * @return the statistics + */ + public Properties getStatistics() { + return statistics; + } + + /** + * Public getter for the commit counter. + * @return the commitCount + */ + public int getCommitCount() { + return commitCount; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepExecution.java index 1868ca120..2a7f9351a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/StepExecution.java @@ -54,7 +54,7 @@ public class StepExecution extends Entity { private Properties statistics = new Properties(); private ExitStatus exitStatus = ExitStatus.UNKNOWN; - + /** * Package private constructor for Hibernate */ @@ -66,11 +66,11 @@ public class StepExecution extends Entity { * Constructor with mandatory properties. * * @param step the step to which this execution belongs - * @param jobExecution the current job execution - */ + * @param jobExecution the current job execution + */ public StepExecution(StepInstance step, JobExecution jobExecution, Long id) { this(); - this.step= step; + this.step = step; this.jobExecution = jobExecution; setId(id); } @@ -87,10 +87,6 @@ public class StepExecution extends Entity { taskCount++; } - public void incrementRollbackCount() { - rollbackCount++; - } - public Properties getStatistics() { return statistics; } @@ -148,7 +144,7 @@ public class StepExecution extends Entity { } public Long getStepId() { - if (step!=null) { + if (step != null) { return step.getId(); } return null; @@ -159,45 +155,48 @@ public class StepExecution extends Entity { * @return the jobExecutionId */ public Long getJobExecutionId() { - if (jobExecution!=null) { + if (jobExecution != null) { return jobExecution.getId(); } return null; } - - /* (non-Javadoc) + + /* + * (non-Javadoc) * @see org.springframework.batch.container.common.domain.Entity#equals(java.lang.Object) */ public boolean equals(Object obj) { Object stepId = getStepId(); Object jobExecutionId = getJobExecutionId(); - if (stepId==null && jobExecutionId==null || !(obj instanceof StepExecution) || getId()!=null) { + if (stepId == null && jobExecutionId == null || !(obj instanceof StepExecution) || getId() != null) { return super.equals(obj); } StepExecution other = (StepExecution) obj; - if (stepId==null) { + if (stepId == null) { return jobExecutionId.equals(other.getJobExecutionId()); } - return stepId.equals(other.getStepId()) && (jobExecutionId==null || jobExecutionId.equals(other.getJobExecutionId())); + return stepId.equals(other.getStepId()) + && (jobExecutionId == null || jobExecutionId.equals(other.getJobExecutionId())); } - - /* (non-Javadoc) + + /* + * (non-Javadoc) * @see org.springframework.batch.container.common.domain.Entity#hashCode() */ public int hashCode() { Object stepId = getStepId(); Object jobExecutionId = getJobExecutionId(); - return super.hashCode() + 31*(stepId!=null ? stepId.hashCode() : 0) + 91*(jobExecutionId!=null ? jobExecutionId.hashCode() : 0); - } - - public String toString() { - return super.toString() + ", name=" + getName() + ", taskCount=" + taskCount + ", commitCount=" + commitCount + ", rollbackCount=" - + rollbackCount; + return super.hashCode() + 31 * (stepId != null ? stepId.hashCode() : 0) + 91 + * (jobExecutionId != null ? jobExecutionId.hashCode() : 0); } + public String toString() { + return super.toString() + ", name=" + getName() + ", taskCount=" + taskCount + ", commitCount=" + commitCount + + ", rollbackCount=" + rollbackCount; + } private String getName() { - return step==null ? null : step.getName(); + return step == null ? null : step.getName(); } /** @@ -231,4 +230,34 @@ public class StepExecution extends Entity { return jobExecution; } + /** + * Factory method for {@link StepContribution}. + * + * @return a new {@link StepContribution} + */ + public StepContribution createStepContribution() { + return new StepContribution(this); + } + + /** + * On successful execution just before a chunk commit, this method should be + * called. Synchronizes access to the {@link StepExecution} so that changes + * are atomic. + * + * @param contribution + */ + public synchronized void apply(StepContribution contribution) { + taskCount += contribution.getTaskCount(); + statistics = contribution.getStatistics(); + commitCount += contribution.getCommitCount(); + } + + /** + * On unsuccessful execution after a chunk has rolled back. Synchronizes + * access to the {@link StepExecution} so that changes are atomic. + */ + public synchronized void rollback() { + rollbackCount++; + } + } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java index ab1387823..67d222c44 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepExecutionTests.java @@ -111,11 +111,11 @@ public class StepExecutionTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.core.domain.StepExecution#incrementRollbackCount()}. + * {@link org.springframework.batch.core.domain.StepExecution#rollback()}. */ public void testIncrementRollbackCount() { int before = execution.getRollbackCount().intValue(); - execution.incrementRollbackCount(); + execution.rollback(); int after = execution.getRollbackCount().intValue(); assertEquals(before + 1, after); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java index 21c1c934e..b16d71b35 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java @@ -185,7 +185,7 @@ public class SimpleJobRepository implements JobRepository { } private JobExecution generateJobExecution(JobInstance job) { - JobExecution execution = job.createNewJobExecution(); + JobExecution execution = job.createJobExecution(); // Save the JobExecution so that it picks up an ID (useful for clients // monitoring asynchronous executions): saveOrUpdate(execution); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlStepDao.java index aced46da8..bd49d2508 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlStepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlStepDao.java @@ -20,7 +20,6 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Types; import java.util.List; -import java.util.Properties; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -36,6 +35,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 +46,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 @@ -66,6 +66,8 @@ public class SqlStepDao implements StepDao, InitializingBean { private static final int EXIT_MESSAGE_LENGTH = 250; + private static final int RESTART_DATA_LENGTH = 1000; + private static final String FIND_STEP = "SELECT ID, STATUS, RESTART_DATA from %PREFIX%STEP where JOB_ID = ? " + "and STEP_NAME = ?"; @@ -89,7 +91,7 @@ public class SqlStepDao 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 +106,7 @@ public class SqlStepDao 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 +123,7 @@ public class SqlStepDao 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 +145,9 @@ public class SqlStepDao 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 +163,28 @@ public class SqlStepDao 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 +194,7 @@ public class SqlStepDao 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 +204,23 @@ public class SqlStepDao 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 +230,7 @@ public class SqlStepDao 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 +242,10 @@ public class SqlStepDao 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 +281,7 @@ public class SqlStepDao implements StepDao, InitializingBean { Object[] parameters = new Object[] { stepId }; - return jdbcTemplate.queryForInt(getStepExecutionCountQuery(), - parameters); + return jdbcTemplate.queryForInt(getStepExecutionCountQuery(), parameters); } private String getStepExecutionCountQuery() { @@ -323,26 +310,15 @@ public class SqlStepDao 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(), + 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.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 SqlStepDao 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 SqlStepDao implements StepDao, InitializingBean { * * @param stepExecutionIncrementer a {@link DataFieldMaxValueIncrementer} */ - public void setStepExecutionIncrementer( - DataFieldMaxValueIncrementer stepExecutionIncrementer) { + public void setStepExecutionIncrementer(DataFieldMaxValueIncrementer stepExecutionIncrementer) { this.stepExecutionIncrementer = stepExecutionIncrementer; } @@ -388,65 +362,65 @@ public class SqlStepDao 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; } /** + * Update the {@link StepExecution}, truncating the exit description. Also + * checks for optimistic locking failure where another agent has updated the + * {@link StepExecution}.
+ * + * N.B. locks the {@link StepExecution} to prevent multi-threaded access. + * + * @throws OptimisticLockingFailureException if the {@link StepExecution} + * version does not match the value in the data base. * @see StepDao#update(StepExecution) */ 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 - // point. - // if (jdbcTemplate.queryForInt(CHECK_STEP_EXECUTION_EXISTS, new - // Object[] { stepExecution.getId() }) != 1) { - // 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) { @@ -454,17 +428,21 @@ public class SqlStepDao implements StepDao, InitializingBean { Assert.notNull(step.getStatus(), "Step status cannot be null."); Assert.notNull(step.getId(), "Step Id cannot be null."); - Properties restartProps = null; + String restartString = ""; RestartData restartData = step.getRestartData(); if (restartData != null) { - restartProps = restartData.getProperties(); + restartString = PropertiesConverter.propertiesToString(restartData.getProperties()); } - Object[] parameters = new Object[] { step.getStatus().toString(), - PropertiesConverter.propertiesToString(restartProps), - step.getId() }; + if (restartString.length() >= RESTART_DATA_LENGTH) { + logger.error("Restart data too long to persist (max length=" + RESTART_DATA_LENGTH + "): " + restartString); + throw new IllegalStateException("Restart exceeded allowed length (" + RESTART_DATA_LENGTH + ")"); + } + + Object[] parameters = new Object[] { step.getStatus().toString(), restartString, step.getId() }; jdbcTemplate.update(getUpdateStepQuery(), parameters); + } /* @@ -476,12 +454,9 @@ public class SqlStepDao 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/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index a53a055ee..4b62de4c9 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -22,6 +22,7 @@ import java.util.Properties; import org.springframework.batch.core.configuration.StepConfiguration; import org.springframework.batch.core.domain.BatchStatus; +import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; import org.springframework.batch.core.executor.ExitCodeExceptionClassifier; @@ -77,12 +78,6 @@ import org.springframework.util.Assert; */ public class SimpleStepExecutor implements StepExecutor { - /** - * Context attribute key for step execution. Used by monitoring and managing - * clients to inspect current step execution. - */ - private static final String STEP_EXECUTION_KEY = "STEP_EXECUTION"; - private RepeatOperations chunkOperations = new RepeatTemplate(); private RepeatOperations stepOperations = new RepeatTemplate(); @@ -97,8 +92,7 @@ public class SimpleStepExecutor implements StepExecutor { // Not for production use... protected PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); - public void setTransactionManager( - PlatformTransactionManager transactionManager) { + public void setTransactionManager(PlatformTransactionManager transactionManager) { this.transactionManager = transactionManager; } @@ -117,8 +111,7 @@ public class SimpleStepExecutor implements StepExecutor { * processing. Should be set up by the caller through a factory. Defaults to * a plain {@link RepeatTemplate}. * - * @param stepOperations - * a {@link RepeatOperations} instance. + * @param stepOperations a {@link RepeatOperations} instance. */ public void setStepOperations(RepeatOperations stepOperations) { this.stepOperations = stepOperations; @@ -129,8 +122,7 @@ public class SimpleStepExecutor implements StepExecutor { * processing. Should be set up by the caller through a factory. Defaults to * a plain {@link RepeatTemplate}. * - * @param chunkOperations - * a {@link RepeatOperations} instance. + * @param chunkOperations a {@link RepeatOperations} instance. */ public void setChunkOperations(RepeatOperations chunkOperations) { this.chunkOperations = chunkOperations; @@ -146,15 +138,13 @@ public class SimpleStepExecutor implements StepExecutor { * execution, which would normally be available to the caller somehow * through the step's {@link JobExecutionContext}.
* - * @throws StepInterruptedException - * if the step or a chunk is interrupted - * @throws RuntimeException - * if there is an exception during a chunk execution + * @throws StepInterruptedException if the step or a chunk is interrupted + * @throws RuntimeException if there is an exception during a chunk + * execution * @see StepExecutor#process(StepConfiguration, StepExecution) */ - public ExitStatus process(final StepConfiguration configuration, - final StepExecution stepExecution) throws BatchCriticalException, - StepInterruptedException { + public ExitStatus process(final StepConfiguration configuration, final StepExecution stepExecution) + throws BatchCriticalException, StepInterruptedException { final StepInstance step = stepExecution.getStep(); boolean isRestart = step.getStepExecutionCount() > 0 ? true : false; @@ -164,17 +154,14 @@ public class SimpleStepExecutor implements StepExecutor { ExitStatus status = ExitStatus.FAILED; - final SimpleStepContext stepScopeContext = StepSynchronizationManager - .open(); + final SimpleStepContext stepScopeContext = StepSynchronizationManager.open(); stepScopeContext.setStepExecution(stepExecution); // Add the job identifier so that it can be used to identify // the conversation in StepScope - stepScopeContext.setAttribute(StepScope.ID_KEY, stepExecution - .getJobExecution().getJob().getIdentifier()); + stepScopeContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getJob().getIdentifier()); try { - stepExecution - .setStartTime(new Timestamp(System.currentTimeMillis())); + stepExecution.setStartTime(new Timestamp(System.currentTimeMillis())); updateStatus(stepExecution, BatchStatus.STARTED); final boolean saveRestartData = configuration.isSaveRestartData(); @@ -185,56 +172,53 @@ public class SimpleStepExecutor implements StepExecutor { status = stepOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) - throws Exception { + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + + final StepContribution contribution = stepExecution.createStepContribution(); + contribution.registerStepContext(context); - stepExecution.getJobExecution() - .registerStepContext(context); - context.registerDestructionCallback( - "STEP_EXECUTION_CONTEXT_CALLBACK", new Runnable() { - public void run() { - stepExecution.getJobExecution() - .unregisterStepContext(context); - } - }); - // Add the step execution as an attribute so monitoring - // clients can see it. - context.setAttribute(STEP_EXECUTION_KEY, stepExecution); // Before starting a new transaction, check for // interruption. interruptionPolicy.checkInterrupted(context); ExitStatus result; + try { - result = (ExitStatus) new TransactionTemplate( - transactionManager) + + result = (ExitStatus) new TransactionTemplate(transactionManager) .execute(new TransactionCallback() { - public Object doInTransaction( - TransactionStatus status) { - // New transaction obtained, - // resynchronize - // TransactionSyncrhonization objects - BatchTransactionSynchronizationManager - .resynchronize(); + public Object doInTransaction(TransactionStatus status) { + /* + * New transaction obtained, + * resynchronize + * TransactionSynchronization objects + */ + BatchTransactionSynchronizationManager.resynchronize(); ExitStatus result; - result = processChunk(configuration, - stepExecution); + result = processChunk(configuration, contribution); + + // TODO: Statistics are not thread safe + // - we cannot guarantee that they are + // up to date. (Maybe we never can?) + Properties statistics = getStatistics(module); + contribution.setStatistics(statistics); + contribution.incrementCommitCount(); + // Apply the contribution to the step + // only if chunk was successful + stepExecution.apply(contribution); if (saveRestartData) { - step - .setRestartData(getRestartData(module)); + step.setRestartData(getRestartData(module)); jobRepository.update(step); } - Properties statistics = getStatistics(module); - stepExecution.setStatistics(statistics); - stepExecution.incrementCommitCount(); - jobRepository - .saveOrUpdate(stepExecution); + jobRepository.saveOrUpdate(stepExecution); return result; } }); - } catch (Throwable t) { + + } + catch (Throwable t) { /* * Any exception thrown within the transaction template * will automatically cause the transaction to rollback. @@ -242,10 +226,11 @@ public class SimpleStepExecutor implements StepExecutor { * commit (e.g. Hibernate flush) so this catch block * comes outside the transaction. */ - stepExecution.incrementRollbackCount(); + stepExecution.rollback(); if (t instanceof RuntimeException) { throw (RuntimeException) t; - } else { + } + else { throw new RuntimeException(t); } } @@ -263,30 +248,34 @@ public class SimpleStepExecutor implements StepExecutor { updateStatus(stepExecution, BatchStatus.COMPLETED); return status; - } catch (RuntimeException e) { + } + catch (RuntimeException e) { // classify exception so an exit code can be stored. status = exceptionClassifier.classifyForExitCode(e); if (e.getCause() instanceof StepInterruptedException) { updateStatus(stepExecution, BatchStatus.STOPPED); throw (StepInterruptedException) e.getCause(); - } else { + } + else { updateStatus(stepExecution, BatchStatus.FAILED); throw e; } - } finally { + } + finally { stepExecution.setExitStatus(status); stepExecution.setEndTime(new Timestamp(System.currentTimeMillis())); try { jobRepository.saveOrUpdate(stepExecution); - } finally { + } + finally { // clear any registered synchronizations try { StepSynchronizationManager.close(); - } finally { - BatchTransactionSynchronizationManager - .clearSynchronizations(); + } + finally { + BatchTransactionSynchronizationManager.clearSynchronizations(); } } } @@ -296,12 +285,9 @@ public class SimpleStepExecutor implements StepExecutor { /** * Convenience method to update the status in all relevant places. * - * @param step - * the current step - * @param stepExecution - * the current stepExecution - * @param status - * the status to set + * @param step the current step + * @param stepExecution the current stepExecution + * @param status the status to set */ private void updateStatus(StepExecution stepExecution, BatchStatus status) { StepInstance step = stepExecution.getStep(); @@ -309,8 +295,7 @@ public class SimpleStepExecutor implements StepExecutor { step.setStatus(status); jobRepository.update(step); jobRepository.saveOrUpdate(stepExecution); - for (Iterator iter = stepExecution.getJobExecution().getStepContexts() - .iterator(); iter.hasNext();) { + for (Iterator iter = stepExecution.getJobExecution().getStepContexts().iterator(); iter.hasNext();) { RepeatContext context = (RepeatContext) iter.next(); context.setAttribute("JOB_STATUS", status); } @@ -322,37 +307,25 @@ public class SimpleStepExecutor implements StepExecutor { * outside this method, so subclasses that override do not need to create a * transaction. * - * @param configuration - * the current step configuration - * @param stepExecution - * the current step, containing the {@link Tasklet} with the - * business logic. + * @param configuration the current step configuration + * @param stepExecution the current step, containing the {@link Tasklet} + * with the business logic. * @return true if there is more data to process. */ - protected final ExitStatus processChunk( - final StepConfiguration configuration, - final StepExecution stepExecution) { - return chunkOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) - throws Exception { - stepExecution.getJobExecution().registerChunkContext(context); - context.registerDestructionCallback( - "CHUNK_EXECUTION_CONTEXT_CALLBACK", new Runnable() { - public void run() { - stepExecution.getJobExecution() - .unregisterStepContext(context); - } - }); + protected final ExitStatus processChunk(final StepConfiguration configuration, final StepContribution contribution) { + ExitStatus result = chunkOperations.iterate(new RepeatCallback() { + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + contribution.registerChunkContext(context); // check for interruption before each item as well interruptionPolicy.checkInterrupted(context); - ExitStatus exitStatus = doTaskletProcessing(configuration - .getTasklet(), stepExecution); - stepExecution.incrementTaskCount(); + ExitStatus exitStatus = doTaskletProcessing(configuration.getTasklet(), contribution); + contribution.incrementTaskCount(); // check for interruption after each item as well interruptionPolicy.checkInterrupted(context); return exitStatus; } }); + return result; } /** @@ -363,23 +336,20 @@ public class SimpleStepExecutor implements StepExecutor { * If there is an exception and the {@link Tasklet} implements * {@link Skippable} then the skip method is called. * - * @param tasklet - * the unit of business logic to execute - * @param stepExecution - * the current step + * @param tasklet the unit of business logic to execute + * @param contribution the current step * @return boolean if there is more processing to do - * @throws Exception - * if there is an error + * @throws Exception if there is an error */ - protected ExitStatus doTaskletProcessing(Tasklet tasklet, - StepExecution stepExecution) throws Exception { + protected ExitStatus doTaskletProcessing(Tasklet tasklet, StepContribution contribution) throws Exception { ExitStatus exitStatus = ExitStatus.CONTINUABLE; try { exitStatus = tasklet.execute(); - } catch (Exception e) { + } + catch (Exception e) { if (tasklet instanceof Skippable) { ((Skippable) tasklet).skip(); @@ -396,12 +366,13 @@ public class SimpleStepExecutor implements StepExecutor { /** * @param tasklet * @return restart data from the {@link Tasklet} if it is - * {@link Restartable} + * {@link Restartable} */ private RestartData getRestartData(Tasklet tasklet) { if (tasklet instanceof Restartable) { return ((Restartable) tasklet).getRestartData(); - } else { + } + else { return null; } } @@ -415,7 +386,8 @@ public class SimpleStepExecutor implements StepExecutor { private Properties getStatistics(Tasklet tasklet) { if (tasklet instanceof StatisticsProvider) { return ((StatisticsProvider) tasklet).getStatistics(); - } else { + } + else { return null; } } @@ -425,8 +397,7 @@ public class SimpleStepExecutor implements StepExecutor { * check whether an external request has been made to interrupt the job * execution. * - * @param interruptionPolicy - * a {@link StepInterruptionPolicy} + * @param interruptionPolicy a {@link StepInterruptionPolicy} */ public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { this.interruptionPolicy = interruptionPolicy; @@ -438,8 +409,7 @@ public class SimpleStepExecutor implements StepExecutor { * * @param exceptionClassifier */ - public void setExceptionClassifier( - ExitCodeExceptionClassifier exceptionClassifier) { + public void setExceptionClassifier(ExitCodeExceptionClassifier exceptionClassifier) { this.exceptionClassifier = exceptionClassifier; } @@ -457,8 +427,7 @@ public class SimpleStepExecutor implements StepExecutor { * {@link SimpleLimitExceptionHandler} with that limit. * * - * @param configuration - * a step configuration + * @param configuration a step configuration */ public void applyConfiguration(StepConfiguration configuration) { @@ -478,27 +447,24 @@ public class SimpleStepExecutor implements StepExecutor { setStepOperations(stepOperations); } - } else if (configuration instanceof SimpleStepConfiguration) { + } + else if (configuration instanceof SimpleStepConfiguration) { SimpleStepConfiguration simpleConfiguation = (SimpleStepConfiguration) configuration; if (this.chunkOperations instanceof RepeatTemplate) { RepeatTemplate template = (RepeatTemplate) this.chunkOperations; - template.setCompletionPolicy(new SimpleCompletionPolicy( - simpleConfiguation.getCommitInterval())); + template.setCompletionPolicy(new SimpleCompletionPolicy(simpleConfiguation.getCommitInterval())); } - ExceptionHandler exceptionHandler = simpleConfiguation - .getExceptionHandler(); + ExceptionHandler exceptionHandler = simpleConfiguation.getExceptionHandler(); - if (simpleConfiguation.getSkipLimit() > 0 - && exceptionHandler == null) { + if (simpleConfiguation.getSkipLimit() > 0 && exceptionHandler == null) { SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler(); handler.setLimit(simpleConfiguation.getSkipLimit()); exceptionHandler = handler; } - if (this.stepOperations instanceof RepeatTemplate - && exceptionHandler != null) { + if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) { RepeatTemplate template = (RepeatTemplate) this.stepOperations; template.setExceptionHandler(exceptionHandler); } diff --git a/spring-batch-execution/src/main/resources/schema-db2.sql b/spring-batch-execution/src/main/resources/schema-db2.sql index 1224021a8..8907e09bb 100644 --- a/spring-batch-execution/src/main/resources/schema-db2.sql +++ b/spring-batch-execution/src/main/resources/schema-db2.sql @@ -35,7 +35,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID BIGINT NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID BIGINT PRIMARY KEY , @@ -47,7 +47,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT BIGINT , TASK_COUNT BIGINT , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-execution/src/main/resources/schema-derby.sql b/spring-batch-execution/src/main/resources/schema-derby.sql index abf9e737c..2ad36f551 100644 --- a/spring-batch-execution/src/main/resources/schema-derby.sql +++ b/spring-batch-execution/src/main/resources/schema-derby.sql @@ -35,7 +35,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID BIGINT NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, @@ -47,7 +47,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT BIGINT , TASK_COUNT BIGINT , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-execution/src/main/resources/schema-hsqldb.sql b/spring-batch-execution/src/main/resources/schema-hsqldb.sql index 43fce576d..9b655fffb 100644 --- a/spring-batch-execution/src/main/resources/schema-hsqldb.sql +++ b/spring-batch-execution/src/main/resources/schema-hsqldb.sql @@ -35,7 +35,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID BIGINT NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID BIGINT IDENTITY PRIMARY KEY , @@ -47,7 +47,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT BIGINT , TASK_COUNT BIGINT , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-execution/src/main/resources/schema-oracle10g.sql b/spring-batch-execution/src/main/resources/schema-oracle10g.sql index 729c7d12d..b8074af09 100644 --- a/spring-batch-execution/src/main/resources/schema-oracle10g.sql +++ b/spring-batch-execution/src/main/resources/schema-oracle10g.sql @@ -35,7 +35,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID NUMBER(38) NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID NUMBER(38) PRIMARY KEY , @@ -47,7 +47,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT NUMBER(38) , TASK_COUNT NUMBER(38) , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-execution/src/main/resources/schema-postgresql.sql b/spring-batch-execution/src/main/resources/schema-postgresql.sql index 1224021a8..8907e09bb 100644 --- a/spring-batch-execution/src/main/resources/schema-postgresql.sql +++ b/spring-batch-execution/src/main/resources/schema-postgresql.sql @@ -35,7 +35,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID BIGINT NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID BIGINT PRIMARY KEY , @@ -47,7 +47,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT BIGINT , TASK_COUNT BIGINT , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-execution/src/main/sql/init.sql.vpp b/spring-batch-execution/src/main/sql/init.sql.vpp index d59bf36af..bb498d609 100644 --- a/spring-batch-execution/src/main/sql/init.sql.vpp +++ b/spring-batch-execution/src/main/sql/init.sql.vpp @@ -24,7 +24,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID ${BIGINT} NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID ${BIGINT} $!{IDENTITY} PRIMARY KEY $!{GENERATED}, @@ -36,7 +36,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT ${BIGINT} , TASK_COUNT ${BIGINT} , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java index 9ce328840..3633f6624 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java @@ -102,8 +102,8 @@ public class SimpleJobRepositoryTests extends TestCase { jobConfiguration.setSteps(stepConfigurations); databaseJob = new JobInstance(jobRuntimeInformation, new Long(1)) { - public JobExecution createNewJobExecution() { - jobExecution = super.createNewJobExecution(); + public JobExecution createJobExecution() { + jobExecution = super.createJobExecution(); return jobExecution; } }; @@ -170,7 +170,7 @@ public class SimpleJobRepositoryTests extends TestCase { jobDaoControl.setReturnValue(1); jobDao.findJobExecutions(databaseJob); final List executions = new ArrayList(); - JobExecution execution =databaseJob.createNewJobExecution(); + JobExecution execution =databaseJob.createJobExecution(); executions.add(execution); // For this test it is important that the execution is finished // and the executions in the list contain one with an end date diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java index df00a5751..8ee3f3762 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java @@ -31,6 +31,7 @@ import org.springframework.batch.execution.runtime.ScheduledJobIdentifier; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.restart.GenericRestartData; import org.springframework.batch.restart.RestartData; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; import org.springframework.util.ClassUtils; @@ -196,7 +197,7 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour StepExecution stepExecution = new StepExecution(null, null); try{ stepDao.update(stepExecution); - fail(); + fail("Expected IllegalArgumentException"); }catch(IllegalArgumentException ex){ //expected } @@ -215,4 +216,25 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour stepDao.save(execution); assertEquals(2, stepDao.getStepExecutionCount(step1.getId())); } + + public void testUpdateStepExecutionVersion() throws Exception { + int before = stepExecution.getVersion().intValue(); + stepDao.update(stepExecution); + int after = stepExecution.getVersion().intValue(); + assertEquals("StepExecution version not updated", before+1, after); + } + + public void testUpdateStepExecutionOptimisticLocking() throws Exception { + stepExecution.incrementVersion(); // not really allowed outside dao code + try { + stepDao.update(stepExecution); + fail("Expected OptimisticLockingFailureException"); + } + catch (OptimisticLockingFailureException e) { + // expected + assertTrue("Exception message should contain step execution id: "+e.getMessage(), e.getMessage().indexOf(""+stepExecution.getId())>=0); + assertTrue("Exception message should contain step execution version: "+e.getMessage(), e.getMessage().indexOf(""+stepExecution.getVersion())>=0); + } + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlJobDaoQueryTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlJobDaoQueryTests.java index c166a8fc1..d4d877a1a 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlJobDaoQueryTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlJobDaoQueryTests.java @@ -66,7 +66,7 @@ public class SqlJobDaoQueryTests extends TestCase { return 1; } }); - sqlDao.save(new JobInstance(new SimpleJobIdentifier("foo"), new Long(11)).createNewJobExecution()); + 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); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlStepDaoTests.java index c0695344e..9b2318543 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlStepDaoTests.java @@ -37,5 +37,5 @@ public class SqlStepDaoTests extends AbstractStepDaoTests { assertEquals(LONG_STRING.substring(0, 250), ((Map) executions.get(0)) .get("EXIT_MESSAGE")); } - + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java index daea2d50c..2a56dbcee 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java @@ -25,6 +25,7 @@ import junit.framework.TestCase; import org.springframework.batch.core.configuration.StepConfigurationSupport; import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobInstance; +import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; import org.springframework.batch.core.executor.ExitCodeExceptionClassifier; @@ -115,6 +116,7 @@ public class DefaultStepExecutorTests extends TestCase { stepExecutor.process(stepConfiguration, stepExecution); assertEquals(1, processed.size()); + assertEquals(1, stepExecution.getTaskCount().intValue()); } public void testChunkExecutor() throws Exception { @@ -131,8 +133,11 @@ public class DefaultStepExecutorTests extends TestCase { jobIdentifier, new Long(1))); StepExecution stepExecution = new StepExecution(step, jobExecution); - stepExecutor.processChunk(stepConfiguration, stepExecution); + StepContribution contribution = stepExecution.createStepContribution(); + stepExecutor.processChunk(stepConfiguration, contribution); assertEquals(1, processed.size()); + assertEquals(0, stepExecution.getTaskCount().intValue()); + assertEquals(1, contribution.getTaskCount()); } diff --git a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql index f3c8be6e8..56aa60fd5 100644 --- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql +++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql @@ -24,7 +24,7 @@ CREATE TABLE BATCH_STEP ( JOB_ID BIGINT NOT NULL, STEP_NAME VARCHAR(100) NOT NULL, STATUS VARCHAR(10), - RESTART_DATA VARCHAR(200)); + RESTART_DATA VARCHAR(1000)); CREATE TABLE BATCH_STEP_EXECUTION ( ID BIGINT IDENTITY PRIMARY KEY , @@ -36,7 +36,7 @@ CREATE TABLE BATCH_STEP_EXECUTION ( STATUS VARCHAR(10), COMMIT_COUNT BIGINT , TASK_COUNT BIGINT , - TASK_STATISTICS VARCHAR(250), + TASK_STATISTICS VARCHAR(1000), CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(250)); diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemReader.java index 41f7ca7b1..2d1beb144 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemReader.java @@ -75,7 +75,9 @@ public class StagingItemReader extends JdbcDaoSupport implements Assert.state(keys == null || initialized, "Cannot open an already open StagingItemProvider" + ", call close() first."); - keys = retrieveKeys().iterator(); + synchronized (lock) { + keys = retrieveKeys().iterator(); + } logger.info("keys: " + keys); registerSynchronization(); initialized = true; @@ -147,7 +149,7 @@ public class StagingItemReader extends JdbcDaoSupport implements throw new OptimisticLockingFailureException( "The staging record with ID=" + id - + " was updated concurrently when trying to mark as complete."); + + " was updated concurrently when trying to mark as complete (updated "+count+" records."); } return result; }