Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - issue BATCH-453: Killed batches cannot be restarted

Browse Source 
http://jira.springframework.org/browse/BATCH-453

The JdbcStepExecutionDao now stores the lastUpdated property of StepExecution into the StepExecution table.
This commit is contained in:
lucasward
2008-08-20 21:30:44 +00:00
parent ef4f2900cc
commit f0b1643c02
5 changed files with 71 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java
View File

@@ -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;
}
}

19
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java
View File

@@ -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;
}

3
spring-batch-core/src/main/resources/schema-hsqldb.sql
View File

@@ -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)
) ;

7
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java
View File

@@ -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
}
}
}

1
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDaoTests.java
View File

@@ -56,4 +56,5 @@ public class JdbcStepExecutionDaoTests extends AbstractStepExecutionDaoTests {
assertTrue("Exit description should be truncated", retrievedAfterUpdate.getExitStatus().getExitDescription()
.length() < stepExecution.getExitStatus().getExitDescription().length());
}
}