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

OPEN - issue BATCH-170: Concurrent modification of StepExecution when running an asynchrounous step operation

Browse Source 
http://opensource.atlassian.com/projects/spring/browse/BATCH-170

Synchronized and added OptimisticLockingException to SqlStepDao.  The parallel job now runs cleanly (tried it a few times).
This commit is contained in:
dsyer
2007-12-19 13:06:31 +00:00
parent 7f71dc55ce
commit 37696b45f4
21 changed files with 431 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
View File

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

217
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlStepDao.java
View File

@@ -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}.<br/>
*
* 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.");
}
}

214
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -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}.<br/>
*
* @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.</li>
* </ul>
*
* @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);
}

4
spring-batch-execution/src/main/resources/schema-db2.sql
View File

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

4
spring-batch-execution/src/main/resources/schema-derby.sql
View File

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

4
spring-batch-execution/src/main/resources/schema-hsqldb.sql
View File

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

4
spring-batch-execution/src/main/resources/schema-oracle10g.sql
View File

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

4
spring-batch-execution/src/main/resources/schema-postgresql.sql
View File

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

4
spring-batch-execution/src/main/sql/init.sql.vpp
View File

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

6
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
View File

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

24
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
View File

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

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlJobDaoQueryTests.java
View File

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

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlStepDaoTests.java
View File

@@ -37,5 +37,5 @@ public class SqlStepDaoTests extends AbstractStepDaoTests {
assertEquals(LONG_STRING.substring(0, 250), ((Map) executions.get(0))
.get("EXIT_MESSAGE"));
}
}

7
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java
View File

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

4
spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql
View File

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