diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java index ef706906f..59c4da654 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java @@ -19,6 +19,8 @@ package org.springframework.batch.core; import java.util.Date; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; import org.springframework.batch.repeat.ExitStatus; import org.springframework.util.Assert; @@ -54,6 +56,8 @@ public class StepExecution extends Entity { private volatile Date startTime = new Date(System.currentTimeMillis()); private volatile Date endTime = null; + + private volatile Date lastUpdated = null; private volatile ExecutionContext executionContext = new ExecutionContext(); @@ -335,14 +339,27 @@ public class StepExecution extends Entity { this.terminateOnly = true; } + /** + * @return the total number of items skipped. + */ public int getSkipCount() { return readSkipCount + writeSkipCount; } + /** + * Increment the number of items skipped in the {@link ItemReader} + * + * @param count - the number of skips to increment by. + */ public void incrementReadSkipCountBy(int count) { readSkipCount += count; } + /** + * Increment the number of items skipped in the {@link ItemWriter} + * + * @param count - the number of skips to increment by. + */ public void incrementWriteSkipCountBy(int count) { writeSkipCount += count; } @@ -360,20 +377,54 @@ public class StepExecution extends Entity { return jobExecution.getJobInstance().getJobParameters(); } + /** + * @return the number of records skipped in the {@link ItemReader} + */ public int getReadSkipCount() { return readSkipCount; } + /** + * @return the number of records skipped in the {@link ItemWriter} + */ public int getWriteSkipCount() { return writeSkipCount; } + /** + * Set the number of records skipped in the {@link ItemReader} + * + * @param readSkipCount + */ public void setReadSkipCount(int readSkipCount) { this.readSkipCount = readSkipCount; } + /** + * Set the number of records skipped in the {@link ItemWriter} + * + * @param writeSkipCount + */ public void setWriteSkipCount(int writeSkipCount) { this.writeSkipCount = writeSkipCount; } + /** + * @return the Date representing the last time this execution was persisted. + */ + public Date getLastUpdated() { + return lastUpdated; + } + + /** + * Set the time when the StepExecution was last updated before + * persisting + * + * @param lastUpdated + */ + public void setLastUpdated(Date lastUpdated) { + this.lastUpdated = lastUpdated; + } + + } 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 80fea843b..916cf456d 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 @@ -41,15 +41,16 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement private static final Log logger = LogFactory.getLog(JdbcStepExecutionDao.class); private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_NAME, JOB_EXECUTION_ID, START_TIME, " - + "END_TIME, STATUS, COMMIT_COUNT, ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT) " - + "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; + + "END_TIME, STATUS, COMMIT_COUNT, ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED) " + + "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, " + "STATUS = ?, COMMIT_COUNT = ?, ITEM_COUNT = ?, CONTINUABLE = ? , EXIT_CODE = ?, " - + "EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ?, ROLLBACK_COUNT = ? where STEP_EXECUTION_ID = ? and VERSION = ?"; + + "EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ?, ROLLBACK_COUNT = ?, LAST_UPDATED = ?" + + " where STEP_EXECUTION_ID = ? and VERSION = ?"; private static final String GET_RAW_STEP_EXECUTIONS = "SELECT STEP_EXECUTION_ID, STEP_NAME, START_TIME, END_TIME, STATUS, COMMIT_COUNT," - + " ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT from %PREFIX%STEP_EXECUTION where JOB_EXECUTION_ID = ?"; + + " ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED from %PREFIX%STEP_EXECUTION where JOB_EXECUTION_ID = ?"; private static final String GET_STEP_EXECUTIONS = GET_RAW_STEP_EXECUTIONS + " order by STEP_EXECUTION_ID"; @@ -106,13 +107,13 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getItemCount(), stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(), exitDescription, stepExecution.getReadSkipCount(), - stepExecution.getWriteSkipCount(), stepExecution.getRollbackCount() }; + stepExecution.getWriteSkipCount(), stepExecution.getRollbackCount(), stepExecution.getLastUpdated() }; getJdbcTemplate().getJdbcOperations().update( getQuery(SAVE_STEP_EXECUTION), parameters, new int[] { Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.CHAR, Types.VARCHAR, - Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER }); + Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.TIMESTAMP }); } /** @@ -154,13 +155,14 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(), exitDescription, version, stepExecution.getReadSkipCount(), stepExecution.getWriteSkipCount(), - stepExecution.getRollbackCount(), stepExecution.getId(), stepExecution.getVersion() }; + stepExecution.getRollbackCount(), stepExecution.getLastUpdated(), stepExecution.getId(), + stepExecution.getVersion() }; int count = getJdbcTemplate().getJdbcOperations().update( getQuery(UPDATE_STEP_EXECUTION), parameters, new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.CHAR, Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER, - Types.INTEGER, Types.INTEGER, Types.INTEGER }); + Types.INTEGER, Types.TIMESTAMP, Types.INTEGER, Types.INTEGER }); // Avoid concurrent modifications... if (count == 0) { @@ -231,6 +233,7 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement stepExecution.setReadSkipCount(rs.getInt(11)); stepExecution.setWriteSkipCount(rs.getInt(12)); stepExecution.setRollbackCount(rs.getInt(13)); + stepExecution.setLastUpdated(rs.getTimestamp(14)); return stepExecution; } diff --git a/spring-batch-core/src/main/resources/schema-hsqldb.sql b/spring-batch-core/src/main/resources/schema-hsqldb.sql index c4f3e8790..174191744 100644 --- a/spring-batch-core/src/main/resources/schema-hsqldb.sql +++ b/spring-batch-core/src/main/resources/schema-hsqldb.sql @@ -59,7 +59,8 @@ CREATE TABLE BATCH_STEP_EXECUTION ( ROLLBACK_COUNT BIGINT , CONTINUABLE CHAR(1) , EXIT_CODE VARCHAR(20) , - EXIT_MESSAGE VARCHAR(2500) , + EXIT_MESSAGE VARCHAR(2500) , + LAST_UPDATED TIMESTAMP , constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID) references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID) ) ; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java index f6a5bf3e3..d0117c25f 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.fail; +import java.util.Date; import java.util.List; import org.junit.Before; @@ -92,10 +93,12 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona @Transactional @Test public void testSaveAndFindExecution() { + stepExecution.setStatus(BatchStatus.STARTED); stepExecution.setReadSkipCount(7); stepExecution.setWriteSkipCount(5); stepExecution.setRollbackCount(3); + stepExecution.setLastUpdated(new Date(System.currentTimeMillis())); dao.saveStepExecution(stepExecution); StepExecution retrieved = dao.getStepExecution(jobExecution, step.getName()); @@ -104,6 +107,7 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona assertEquals(stepExecution.getReadSkipCount(), retrieved.getReadSkipCount()); assertEquals(stepExecution.getWriteSkipCount(), retrieved.getWriteSkipCount()); assertEquals(stepExecution.getRollbackCount(), retrieved.getRollbackCount()); + assertEquals(stepExecution.getLastUpdated(), retrieved.getLastUpdated()); assertNull(dao.getStepExecution(jobExecution, "not-existing step")); } @@ -167,11 +171,13 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona Integer versionAfterSave = stepExecution.getVersion(); stepExecution.setStatus(BatchStatus.STOPPED); + stepExecution.setLastUpdated(new Date(System.currentTimeMillis())); dao.updateStepExecution(stepExecution); assertEquals(versionAfterSave + 1, stepExecution.getVersion().intValue()); StepExecution retrieved = dao.getStepExecution(jobExecution, step.getName()); assertEquals(stepExecution, retrieved); + assertEquals(stepExecution.getLastUpdated(), retrieved.getLastUpdated()); assertEquals(BatchStatus.STOPPED, retrieved.getStatus()); } @@ -205,5 +211,4 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona } } - } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDaoTests.java index 2fa9ce0be..6bac7d2f3 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDaoTests.java @@ -56,4 +56,5 @@ public class JdbcStepExecutionDaoTests extends AbstractStepExecutionDaoTests { assertTrue("Exit description should be truncated", retrievedAfterUpdate.getExitStatus().getExitDescription() .length() < stepExecution.getExitStatus().getExitDescription().length()); } + }