First whack at removing the ItemOrientedTasklet. I realize this doesn't yet compile to pass tests.

This commit is contained in:
nebhale
2008-02-09 17:32:39 +00:00
parent 012e2634f8
commit 75ff70d027
10 changed files with 425 additions and 531 deletions

View File

@@ -20,11 +20,14 @@ import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInterruptedException;
import org.springframework.batch.core.domain.StepSupport;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.tasklet.Tasklet;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.stream.SimpleStreamManager;
import org.springframework.batch.item.stream.StreamManager;
import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
@@ -32,9 +35,9 @@ import org.springframework.util.Assert;
* A {@link Step} implementation that provides common behaviour to subclasses.
*
* @author Dave Syer
*
* @author Ben Hale
*/
public abstract class AbstractStep extends StepSupport {
public abstract class AbstractStep extends StepSupport implements InitializingBean {
private int skipLimit = 0;
@@ -44,10 +47,14 @@ public abstract class AbstractStep extends StepSupport {
private PlatformTransactionManager transactionManager;
private Tasklet tasklet;
private StreamManager streamManager;
private ItemReader itemReader;
private ItemWriter itemWriter;
private ItemRecoverer itemRecoverer;
/**
* Default constructor.
*/
@@ -57,6 +64,7 @@ public abstract class AbstractStep extends StepSupport {
/**
* Convenient constructor for setting only the name property.
*
* @param name
*/
public AbstractStep(String name) {
@@ -90,6 +98,7 @@ public abstract class AbstractStep extends StepSupport {
/**
* Public setter for the {@link PlatformTransactionManager}.
*
* @param transactionManager the transaction manager to set
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
@@ -97,14 +106,35 @@ public abstract class AbstractStep extends StepSupport {
}
/**
* Public setter for the {@link StreamManager}. Set either this or the
* transaction manager, but not both.
* Public setter for the {@link StreamManager}. Set either this or the transaction manager, but not both.
*
* @param streamManager the {@link StreamManager} to set.
*/
public void setStreamManager(StreamManager streamManager) {
this.streamManager = streamManager;
}
/**
* @param itemReader the itemReader to set
*/
public void setItemReader(ItemReader itemReader) {
this.itemReader = itemReader;
}
/**
* @param itemWriter the itemWriter to set
*/
public void setItemWriter(ItemWriter itemWriter) {
this.itemWriter = itemWriter;
}
/**
* @param itemRecoverer the itemRecoverer to set
*/
public void setItemRecoverer(ItemRecoverer itemRecoverer) {
this.itemRecoverer = itemRecoverer;
}
/**
* Assert that all mandatory properties are set (the {@link JobRepository}).
*
@@ -117,33 +147,24 @@ public abstract class AbstractStep extends StepSupport {
protected void assertMandatoryProperties() {
Assert.notNull(jobRepository, "JobRepository is mandatory");
Assert.state(transactionManager != null || streamManager != null,
"Either StreamManager or TransactionManager must be set");
"Either StreamManager or TransactionManager must be set");
Assert.state(transactionManager == null || streamManager == null,
"Only one of StreamManager or TransactionManager must be set");
"Only one of StreamManager or TransactionManager must be set");
Assert.notNull(itemReader, "ItemReader must be provided");
Assert.notNull(itemWriter, "ItemWriter must be provided");
}
/*
* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepSupport#process(org.springframework.batch.core.domain.StepExecution)
*/
public void execute(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException {
SimpleStepExecutor executor = createStepExecutor();
executor.execute(stepExecution);
}
/**
* Public setter for the tasklet.
*
* @param tasklet the tasklet to set
*/
public void setTasklet(Tasklet tasklet) {
this.tasklet = tasklet;
}
/**
* @return a {@link SimpleStepExecutor} that can be used to launch the job.
* @throws BatchCriticalException
*/
protected SimpleStepExecutor createStepExecutor() {
protected SimpleStepExecutor createStepExecutor() throws BatchCriticalException {
assertMandatoryProperties();
// Do not set the streamManager field if it is null, otherwise
// the mandatory properties check will fail.
@@ -152,11 +173,17 @@ public abstract class AbstractStep extends StepSupport {
manager = new SimpleStreamManager(transactionManager);
}
SimpleStepExecutor executor = new SimpleStepExecutor(this);
executor.setItemReader(itemReader);
executor.setItemWriter(itemWriter);
executor.setItemRecoverer(itemRecoverer);
executor.setRepository(jobRepository);
executor.setStreamManager(manager);
try {
executor.afterPropertiesSet();
} catch (Exception e) {
throw new BatchCriticalException(e);
}
executor.applyConfiguration(this);
executor.setTasklet(tasklet);
return executor;
}
}

View File

@@ -19,6 +19,7 @@ package org.springframework.batch.execution.step.simple;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInterruptedException;
import org.springframework.batch.core.tasklet.Tasklet;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.repeat.RepeatOperations;
@@ -28,15 +29,15 @@ import org.springframework.batch.repeat.RepeatOperations;
*
* @author Lucas Ward
* @author Dave Syer
*
* @author Ben Hale
*/
public class RepeatOperationsStep extends AbstractStep implements RepeatOperationsHolder {
// default chunkOperations is null
private RepeatOperations chunkOperations;
private volatile RepeatOperations chunkOperations;
// default stepOperations is null
private RepeatOperations stepOperations;
private volatile RepeatOperations stepOperations;
private volatile Tasklet tasklet;
/**
* Public accessor for the chunkOperations property.
@@ -55,6 +56,11 @@ public class RepeatOperationsStep extends AbstractStep implements RepeatOperatio
public void setChunkOperations(RepeatOperations chunkOperations) {
this.chunkOperations = chunkOperations;
}
public void setTasklet(Tasklet tasklet) {
this.tasklet = tasklet;
}
/**
* Public accessor for the stepOperations property.
@@ -74,9 +80,6 @@ public class RepeatOperationsStep extends AbstractStep implements RepeatOperatio
this.stepOperations = stepOperations;
}
/* (non-Javadoc)
* @see org.springframework.batch.execution.step.simple.AbstractStep#process(org.springframework.batch.core.domain.StepExecution)
*/
public void execute(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException {
assertMandatoryProperties();
SimpleStepExecutor executor = (SimpleStepExecutor) super.createStepExecutor();

View File

@@ -17,20 +17,20 @@
package org.springframework.batch.execution.step.simple;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.tasklet.Tasklet;
/**
* Simple {@link Step} good enough for most purposes and easy to
* configure simple properties, principally the commit interval.
* Simple {@link Step} good enough for most purposes and easy to configure simple properties, principally the commit
* interval.
*
* @author Lucas Ward
* @author Dave Syer
*
* @author Ben Hale
*/
public class SimpleStep extends AbstractStep {
// default commit interval is one
private int commitInterval = 1;
public SimpleStep() {
super();
}
@@ -39,11 +39,6 @@ public class SimpleStep extends AbstractStep {
super(name);
}
public SimpleStep(Tasklet module) {
this();
setTasklet(module);
}
public void setCommitInterval(int commitInterval) {
this.commitInterval = commitInterval;
}

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.execution.step.simple;
import java.util.Date;
@@ -36,7 +35,11 @@ import org.springframework.batch.execution.scope.StepSynchronizationManager;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.stream.SimpleStreamManager;
import org.springframework.batch.item.stream.StreamManager;
@@ -48,30 +51,31 @@ import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.Assert;
/**
* Simple implementation of executing the step as a set of chunks, each chunk
* surrounded by a transaction. The structure is therefore that of two nested
* loops, with transaction boundary around the whole inner loop. The outer loop
* is controlled by the step operations ({@link #setStepOperations(RepeatOperations)}),
* and the inner loop by the chunk operations ({@link #setChunkOperations(RepeatOperations)}).
* The inner loop should always be executed in a single thread, so the chunk
* operations should not do any concurrent execution. N.B. usually that means
* that the chunk operations should be a {@link RepeatTemplate} (which is the
* default).<br/>
* Simple implementation of executing the step as a set of chunks, each chunk surrounded by a transaction. The structure
* is therefore that of two nested loops, with transaction boundary around the whole inner loop. The outer loop is
* controlled by the step operations ({@link #setStepOperations(RepeatOperations)}), and the inner loop by the chunk
* operations ({@link #setChunkOperations(RepeatOperations)}). The inner loop should always be executed in a single
* thread, so the chunk operations should not do any concurrent execution. N.B. usually that means that the chunk
* operations should be a {@link RepeatTemplate} (which is the default).<br/>
*
* Clients can use interceptors in the step operations to intercept or listen to
* the iteration on a step-wide basis, for instance to get a callback when the
* step is complete. Those that want callbacks at the level of an individual
* tasks, can specify interceptors for the chunk operations.
* Clients can use interceptors in the step operations to intercept or listen to the iteration on a step-wide basis, for
* instance to get a callback when the step is complete. Those that want callbacks at the level of an individual tasks,
* can specify interceptors for the chunk operations.
*
* @author Dave Syer
* @author Lucas Ward
*
* @author Ben Hale
*/
public class SimpleStepExecutor {
public class SimpleStepExecutor implements InitializingBean {
private static final Log logger = LogFactory.getLog(SimpleStepExecutor.class);
@@ -86,12 +90,22 @@ public class SimpleStepExecutor {
// default to checking current thread for interruption.
private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy();
private Tasklet tasklet;
private AbstractStep step;
private StreamManager streamManager;
private ItemReader itemReader;
private ItemWriter itemWriter;
private RetryPolicy retryPolicy = null;
private ItemRecoverer itemRecoverer;
private RetryTemplate template = new RetryTemplate();
private ItemReaderRetryCallback retryCallback;
/**
* Package private constructor so the step can create a the executor.
*/
@@ -100,22 +114,18 @@ public class SimpleStepExecutor {
}
/**
* Public setter for the {@link StreamManager}. This will be used to create
* the {@link StepContext}, and hence any component that is a
* {@link ItemStream} and in step scope will be registered with the service.
* The {@link StepContext} is then a source of aggregate statistics for the
* step.
* Public setter for the {@link StreamManager}. This will be used to create the {@link StepContext}, and hence any
* component that is a {@link ItemStream} and in step scope will be registered with the service. The
* {@link StepContext} is then a source of aggregate statistics for the step.
*
* @param streamManager the {@link StreamManager} to set. Default is a
* {@link SimpleStreamManager}.
* @param streamManager the {@link StreamManager} to set. Default is a {@link SimpleStreamManager}.
*/
public void setStreamManager(StreamManager streamManager) {
this.streamManager = streamManager;
}
/**
* Injected strategy for storage and retrieval of persistent step
* information. Mandatory property.
* Injected strategy for storage and retrieval of persistent step information. Mandatory property.
*
* @param jobRepository
*/
@@ -124,9 +134,8 @@ public class SimpleStepExecutor {
}
/**
* The {@link RepeatOperations} to use for the outer loop of the batch
* processing. Should be set up by the caller through a factory. Defaults to
* a plain {@link RepeatTemplate}.
* The {@link RepeatOperations} to use for the outer loop of the batch processing. Should be set up by the caller
* through a factory. Defaults to a plain {@link RepeatTemplate}.
*
* @param stepOperations a {@link RepeatOperations} instance.
*/
@@ -135,9 +144,8 @@ public class SimpleStepExecutor {
}
/**
* The {@link RepeatOperations} to use for the inner loop of the batch
* processing. Should be set up by the caller through a factory. Defaults to
* a plain {@link RepeatTemplate}.
* The {@link RepeatOperations} to use for the inner loop of the batch processing. Should be set up by the caller
* through a factory. Defaults to a plain {@link RepeatTemplate}.
*
* @param chunkOperations a {@link RepeatOperations} instance.
*/
@@ -146,18 +154,121 @@ public class SimpleStepExecutor {
}
/**
* Process the step and update its context so that progress can be monitored
* by the caller. The step is broken down into chunks, each one executing in
* a transaction. The step and its execution and execution context are all
* given an up to date {@link BatchStatus}, and the {@link JobRepository}
* is used to store the result. Various reporting information are also added
* to the current context (the {@link RepeatContext} governing the step
* execution, which would normally be available to the caller somehow
* through the step's {@link JobExecutionContext}.<br/>
* Setter for the {@link StepInterruptionPolicy}. The policy is used to check whether an external request has been
* made to interrupt the job execution.
*
* @param interruptionPolicy a {@link StepInterruptionPolicy}
*/
public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) {
this.interruptionPolicy = interruptionPolicy;
}
/**
* Setter for the {@link ExitCodeExceptionClassifier} that will be used to classify any exception that causes a job
* to fail.
*
* @param exceptionClassifier
*/
public void setExceptionClassifier(ExitCodeExceptionClassifier exceptionClassifier) {
this.exceptionClassifier = exceptionClassifier;
}
/**
* @param itemReader
*/
public void setItemReader(ItemReader itemReader) {
this.itemReader = itemReader;
}
/**
* @param itemWriter
*/
public void setItemWriter(ItemWriter itemWriter) {
this.itemWriter = itemWriter;
}
/**
* Setter for injecting optional recovery handler.
*
* @param itemRecoverer
*/
public void setItemRecoverer(ItemRecoverer itemRecoverer) {
this.itemRecoverer = itemRecoverer;
}
/**
* Public setter for the retryPolicy.
*
* @param retyPolicy the retryPolicy to set
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* Check mandatory properties (reader and writer).
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(itemReader, "ItemReader must be provided");
Assert.notNull(itemWriter, "ItemWriter must be provided");
if (itemRecoverer == null && (itemReader instanceof ItemRecoverer)) {
itemRecoverer = (ItemRecoverer) itemReader;
}
ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy);
template.setRetryPolicy(itemProviderRetryPolicy);
if (retryPolicy != null) {
Assert.state(itemReader instanceof KeyedItemReader,
"ItemReader must be instance of KeyedItemReader to use the retry policy");
retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter);
retryCallback.setRecoverer(itemRecoverer);
}
}
/**
* Apply the configuration by inspecting it to see if it has any relevant policy information.
*
* @param step a step
*/
void applyConfiguration(AbstractStep step) {
if (step instanceof SimpleStep) {
SimpleStep simple = (SimpleStep) step;
if (this.chunkOperations instanceof RepeatTemplate) {
RepeatTemplate template = (RepeatTemplate) this.chunkOperations;
template.setCompletionPolicy(new SimpleCompletionPolicy(simple.getCommitInterval()));
}
}
ExceptionHandler exceptionHandler = step.getExceptionHandler();
if (step.getSkipLimit() > 0 && exceptionHandler == null) {
SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler();
handler.setLimit(step.getSkipLimit());
exceptionHandler = handler;
}
if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) {
RepeatTemplate template = (RepeatTemplate) this.stepOperations;
template.setExceptionHandler(exceptionHandler);
}
}
/**
* Process the step and update its context so that progress can be monitored by the caller. The step is broken down
* into chunks, each one executing in a transaction. The step and its execution and execution context are all given
* an up to date {@link BatchStatus}, and the {@link JobRepository} is used to store the result. Various reporting
* information are also added to the current context (the {@link RepeatContext} governing the step 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 RuntimeException if there is an exception during a chunk execution
* @see StepExecutor#execute(StepExecution)
*/
public void execute(final StepExecution stepExecution) throws BatchCriticalException, StepInterruptedException {
@@ -205,8 +316,7 @@ public class SimpleStepExecutor {
try {
/*
* New transaction obtained, resynchronize
* TransactionSynchronization objects
* New transaction obtained, resynchronize TransactionSynchronization objects
*/
result = processChunk(step, contribution);
@@ -235,38 +345,31 @@ public class SimpleStepExecutor {
streamManager.commit(transaction);
}
catch (Throwable t) {
} catch (Throwable t) {
/*
* Any exception thrown within the transaction template
* will automatically cause the transaction to rollback.
* We need to include exceptions during an attempted
* commit (e.g. Hibernate flush) so this catch block
* comes outside the transaction.
* Any exception thrown within the transaction template will automatically cause the transaction
* to rollback. We need to include exceptions during an attempted commit (e.g. Hibernate flush)
* so this catch block comes outside the transaction.
*/
synchronized (stepExecution) {
stepExecution.rollback();
}
try {
streamManager.rollback(transaction);
}
catch (ResetFailedException e) {
} catch (ResetFailedException e) {
// The original Throwable cause is in danger of
// being lost here, so we log the reset
// failure and re-throw with cause of the rollback.
logger.error("Encountered reset error on rollback: "
+ "one of the streams may be in an inconsistent state, "
+ "so this step should not proceed", e);
throw new ResetFailedException(
"Encountered reset error on rollback. " +
"Consult logs for the cause of the reet failure. " +
"The cause of the original rollback is incuded here.",
t);
+ "one of the streams may be in an inconsistent state, "
+ "so this step should not proceed", e);
throw new ResetFailedException("Encountered reset error on rollback. "
+ "Consult logs for the cause of the reet failure. "
+ "The cause of the original rollback is incuded here.", t);
}
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
}
else {
} else {
throw new RuntimeException(t);
}
}
@@ -282,32 +385,27 @@ public class SimpleStepExecutor {
});
updateStatus(stepExecution, BatchStatus.COMPLETED);
}
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 if (e instanceof ResetFailedException) {
} else if (e instanceof ResetFailedException) {
updateStatus(stepExecution, BatchStatus.UNKNOWN);
throw (ResetFailedException) e;
}
else {
} else {
updateStatus(stepExecution, BatchStatus.FAILED);
throw e;
}
}
finally {
} finally {
stepExecution.setExitStatus(status);
stepExecution.setEndTime(new Date(System.currentTimeMillis()));
try {
jobRepository.saveOrUpdate(stepExecution);
}
finally {
} finally {
// clear any registered synchronizations
StepSynchronizationManager.close();
}
@@ -315,6 +413,116 @@ public class SimpleStepExecutor {
}
/**
* Execute a bunch of identical business logic operations all within a transaction. The transaction is
* programmatically started and stopped outside this method, so subclasses that override do not need to create a
* transaction.
*
* @param step the current step containing the {@link Tasklet} with the business logic.
* @return true if there is more data to process.
*/
ExitStatus processChunk(final Step step, final StepContribution contribution) {
ExitStatus result = chunkOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
if (contribution.isTerminateOnly()) {
context.setTerminateOnly();
}
// check for interruption before each item as well
interruptionPolicy.checkInterrupted(context);
ExitStatus exitStatus = doProcessing(contribution);
contribution.incrementTaskCount();
// check for interruption after each item as well
interruptionPolicy.checkInterrupted(context);
return exitStatus;
}
});
return result;
}
/**
* Execute the business logic, delegating to the given {@link Tasklet}. Subclasses could extend the behaviour as
* long as they always return the value of this method call in their superclass.<br/>
*
* 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 contribution the current step
* @return boolean if there is more processing to do
* @throws Exception if there is an error
*/
private ExitStatus doProcessing(StepContribution contribution) throws Exception {
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
try {
exitStatus = execute();
} catch (Exception e) {
skip();
// Rethrow so that outer transaction is rolled back properly
throw e;
}
return exitStatus;
}
/**
* Read from the {@link ItemReader} and process (if not null) with the {@link ItemWriter}. The call to
* {@link ItemWriter} is wrapped in a stateful retry, if a {@link RetryPolicy} is provided. The
* {@link ItemRecoverer} is used (if provided) in the case of an exception to apply alternate processing to the
* item. If the stateful retry is in place then the recovery will happen in the next transaction automatically,
* otherwise it might be necessary for clients to make the recover method transactional with appropriate propagation
* behaviour (probably REQUIRES_NEW because the call will happen in the context of a transaction that is about to
* rollback).
*
* @see org.springframework.batch.core.tasklet.Tasklet#execute()
*/
private ExitStatus execute() throws Exception {
if (retryCallback == null) {
Object item = itemReader.read();
if (item == null) {
return ExitStatus.FINISHED;
}
try {
itemWriter.write(item);
} catch (Exception e) {
if (itemRecoverer != null) {
itemRecoverer.recover(item, e);
}
// Re-throw the exception so that the surrounding transaction
// rolls back if there is one
throw e;
}
return ExitStatus.CONTINUABLE;
}
return new ExitStatus(template.execute(retryCallback) != null);
}
/**
* Mark the current item as skipped if possible. If there is a retry policy in action there is no need to take any
* action now because it will be covered by the retry in the next transaction. Otherwise if the reader and / or
* writer are {@link Skippable} then delegate to them in that order.
*
* @see org.springframework.batch.io.Skippable#skip()
*/
private void skip() {
if (retryCallback != null) {
// No need to skip because the recoverer will take any action
// necessary.
return;
}
if (this.itemReader instanceof Skippable) {
((Skippable) this.itemReader).skip();
}
if (this.itemWriter instanceof Skippable) {
((Skippable) this.itemWriter).skip();
}
}
/**
* Convenience method to update the status in all relevant places.
*
@@ -328,126 +536,4 @@ public class SimpleStepExecutor {
jobRepository.update(step);
jobRepository.saveOrUpdate(stepExecution);
}
/**
* Execute a bunch of identical business logic operations all within a
* transaction. The transaction is programmatically started and stopped
* outside this method, so subclasses that override do not need to create a
* transaction.
*
* @param step 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 Step step, final StepContribution contribution) {
ExitStatus result = chunkOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
if (contribution.isTerminateOnly()) {
context.setTerminateOnly();
}
// check for interruption before each item as well
interruptionPolicy.checkInterrupted(context);
ExitStatus exitStatus = doTaskletProcessing(tasklet, contribution);
contribution.incrementTaskCount();
// check for interruption after each item as well
interruptionPolicy.checkInterrupted(context);
return exitStatus;
}
});
return result;
}
/**
* Execute the business logic, delegating to the given {@link Tasklet}.
* Subclasses could extend the behaviour as long as they always return the
* value of this method call in their superclass.<br/>
*
* 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 contribution the current step
* @return boolean if there is more processing to do
* @throws Exception if there is an error
*/
protected ExitStatus doTaskletProcessing(Tasklet tasklet, StepContribution contribution) throws Exception {
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
try {
exitStatus = tasklet.execute();
}
catch (Exception e) {
if (tasklet instanceof Skippable) {
((Skippable) tasklet).skip();
}
// Rethrow so that outer transaction is rolled back properly
throw e;
}
return exitStatus;
}
/**
* Setter for the {@link StepInterruptionPolicy}. The policy is used to
* check whether an external request has been made to interrupt the job
* execution.
*
* @param interruptionPolicy a {@link StepInterruptionPolicy}
*/
public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) {
this.interruptionPolicy = interruptionPolicy;
}
/**
* Setter for the {@link ExitCodeExceptionClassifier} that will be used to
* classify any exception that causes a job to fail.
*
* @param exceptionClassifier
*/
public void setExceptionClassifier(ExitCodeExceptionClassifier exceptionClassifier) {
this.exceptionClassifier = exceptionClassifier;
}
/**
* Apply the configuration by inspecting it to see if it has any relevant
* policy information.
*
* @param step a step
*/
void applyConfiguration(AbstractStep step) {
if (step instanceof SimpleStep) {
SimpleStep simple = (SimpleStep) step;
if (this.chunkOperations instanceof RepeatTemplate) {
RepeatTemplate template = (RepeatTemplate) this.chunkOperations;
template.setCompletionPolicy(new SimpleCompletionPolicy(simple.getCommitInterval()));
}
}
ExceptionHandler exceptionHandler = step.getExceptionHandler();
if (step.getSkipLimit() > 0 && exceptionHandler == null) {
SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler();
handler.setLimit(step.getSkipLimit());
exceptionHandler = handler;
}
if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) {
RepeatTemplate template = (RepeatTemplate) this.stepOperations;
template.setExceptionHandler(exceptionHandler);
}
}
/**
* @param tasklet a {@link Tasklet} to execute when a step is processed
*/
public void setTasklet(Tasklet tasklet) {
this.tasklet = tasklet;
}
}

View File

@@ -1,215 +0,0 @@
/*
* 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.execution.tasklet;
import org.springframework.batch.core.tasklet.Tasklet;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
/**
* A concrete implementation of the {@link Tasklet} interface that provides
* 'split processing'. This type of processing is characterized by separating
* the reading and processing of batch data into two separate classes:
* {@link ItemReader} and {@link ItemWriter}. The {@link ItemReader} class
* provides a solid means for re-usability and enforces good architecture
* practices. Because an object <em>must</em> be returned by the
* {@link ItemReader} to continue processing, (returning null indicates
* processing should end) a developer is forced to read in all relevant data,
* place it into a domain object, and return that object. The
* {@link ItemWriter} will then use this object for calculations and output.<br/>
*
* If a {@link RetryPolicy} is provided it will be used to construct a stateful
* retry around the {@link ItemWriter}, delegating identity concerns to the
* {@link ItemReader} and recovery concerns to the {@link ItemRecoverer} (if
* present). In this case clients of this class do not need to take any
* additional action at runtime to take advantage of the retry and recovery,
* provided that when the {@link #execute()} method is called again the same
* item is eventually re-presented (normally this would be the case because a
* transaction would have rolled back and the {@link ItemReader} would go back
* to its previous state).<br/>
*
* If a {@link RetryPolicy} is not provided then the {@link ItemRecoverer} can
* be used to attempt to recover immediately (with no retry) from a processing
* error. Clients of this class should ensure that the recovery takes place in a
* separate transaction (e.g. with propagation REQUIRES_NEW) if necessary. This
* can be achieved by injecting an {@link ItemRecoverer} that has a
* transactional recover method.
*
* @see ItemReader
* @see ItemWriter
* @see RetryPolicy
* @see Recoverable
*
* @author Lucas Ward
* @author Dave Syer
* @author Robert Kasanicky
*
*/
public class ItemOrientedTasklet implements Tasklet, Skippable, InitializingBean {
/**
* Prefix added to statistics keys from writer if needed to avoid
* ambiguity between reader and writer.
*/
public static final String WRITER_STATISTICS_PREFIX = "writer.";
/**
* Prefix added to statistics keys from reader if needed to avoid
* ambiguity between provider and writer.
*/
public static final String READER_STATISTICS_PREFIX = "reader.";
private RetryPolicy retryPolicy = null;
protected ItemReader itemReader;
protected ItemWriter itemWriter;
private ItemRecoverer itemRecoverer;
private RetryTemplate template = new RetryTemplate();
private ItemReaderRetryCallback retryCallback;
/**
* Check mandatory properties (reader and writer).
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(itemReader, "ItemReader must be provided");
Assert.notNull(itemWriter, "ItemWriter must be provided");
if (itemRecoverer == null && (itemReader instanceof ItemRecoverer)) {
itemRecoverer = (ItemRecoverer) itemReader;
}
ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy);
template.setRetryPolicy(itemProviderRetryPolicy);
if (retryPolicy != null) {
Assert.state(itemReader instanceof KeyedItemReader, "ItemReader must be instance of KeyedItemReader to use the retry policy");
retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter);
retryCallback.setRecoverer(itemRecoverer);
}
}
/**
* Read from the {@link ItemReader} and process (if not null) with the
* {@link ItemWriter}. The call to {@link ItemWriter} is wrapped in a
* stateful retry, if a {@link RetryPolicy} is provided. The
* {@link ItemRecoverer} is used (if provided) in the case of an exception
* to apply alternate processing to the item. If the stateful retry is in
* place then the recovery will happen in the next transaction
* automatically, otherwise it might be necessary for clients to make the
* recover method transactional with appropriate propagation behaviour
* (probably REQUIRES_NEW because the call will happen in the context of a
* transaction that is about to rollback).
*
* @see org.springframework.batch.core.tasklet.Tasklet#execute()
*/
public ExitStatus execute() throws Exception {
if (retryCallback == null) {
Object item = itemReader.read();
if (item == null) {
return ExitStatus.FINISHED;
}
try {
itemWriter.write(item);
}
catch (Exception e) {
if (itemRecoverer != null) {
itemRecoverer.recover(item, e);
}
// Re-throw the exception so that the surrounding transaction
// rolls back if there is one
throw e;
}
return ExitStatus.CONTINUABLE;
}
return new ExitStatus(template.execute(retryCallback) != null);
}
/**
* @param itemReader
*/
public void setItemReader(ItemReader itemReader) {
this.itemReader = itemReader;
}
/**
* @param itemWriter
*/
public void setItemWriter(ItemWriter itemWriter) {
this.itemWriter = itemWriter;
}
/**
* Setter for injecting optional recovery handler.
*
* @param itemRecoverer
*/
public void setItemRecoverer(ItemRecoverer itemRecoverer) {
this.itemRecoverer = itemRecoverer;
}
/**
* Mark the current item as skipped if possible. If there is a retry policy
* in action there is no need to take any action now because it will be
* covered by the retry in the next transaction. Otherwise if the reader
* and / or writer are {@link Skippable} then delegate to them in that
* order.
*
* @see org.springframework.batch.io.Skippable#skip()
*/
public void skip() {
if (retryCallback != null) {
// No need to skip because the recoverer will take any action
// necessary.
return;
}
if (this.itemReader instanceof Skippable) {
((Skippable) this.itemReader).skip();
}
if (this.itemWriter instanceof Skippable) {
((Skippable) this.itemWriter).skip();
}
}
/**
* Public setter for the retryPolicy.
*
* @param retyPolicy the retryPolicy to set
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
}

View File

@@ -251,6 +251,7 @@ public class SimpleJobTests extends TestCase {
private Runnable runnable;
private Exception exception;
private Tasklet tasklet;
/**
* @param string
@@ -265,6 +266,13 @@ public class SimpleJobTests extends TestCase {
public void setProcessException(Exception exception) {
this.exception = exception;
}
/**
* @param tasklet the tasklet to set
*/
public void setTasklet(Tasklet tasklet) {
this.tasklet = tasklet;
}
/**
* @param runnable

View File

@@ -27,15 +27,12 @@ import org.springframework.batch.core.domain.BatchStatus;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.tasklet.Tasklet;
import org.springframework.batch.execution.job.simple.SimpleJob;
import org.springframework.batch.execution.repository.SimpleJobRepository;
import org.springframework.batch.execution.repository.dao.MapJobDao;
import org.springframework.batch.execution.repository.dao.MapStepDao;
import org.springframework.batch.execution.step.simple.AbstractStep;
import org.springframework.batch.execution.step.simple.RepeatOperationsStep;
import org.springframework.batch.execution.step.simple.SimpleStep;
import org.springframework.batch.execution.tasklet.ItemOrientedTasklet;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
@@ -71,42 +68,40 @@ public class SimpleJobTests extends TestCase {
job.setJobRepository(repository);
}
private Tasklet getTasklet(String arg) throws Exception {
return getTasklet(new String[] { arg });
private AbstractStep getStep(String arg) throws Exception {
return getStep(new String[] { arg });
}
private Tasklet getTasklet(String arg0, String arg1) throws Exception {
return getTasklet(new String[] { arg0, arg1 });
private AbstractStep getStep(String arg0, String arg1) throws Exception {
return getStep(new String[] { arg0, arg1 });
}
private ItemOrientedTasklet getTasklet(String[] args) throws Exception {
ItemOrientedTasklet module = new ItemOrientedTasklet();
private AbstractStep getStep(String[] args) throws Exception {
SimpleStep step = new SimpleStep();
List items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(args));
provider = new ListItemReader(items);
module.setItemRecoverer(new ItemRecoverer() {
step.setItemRecoverer(new ItemRecoverer() {
public boolean recover(Object item, Throwable cause) {
recovered.add(item);
assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
return true;
}
});
module.setItemReader(provider);
module.setItemWriter(processor);
module.afterPropertiesSet();
return module;
step.setItemReader(provider);
step.setItemWriter(processor);
step.setJobRepository(repository);
step.setTransactionManager(new ResourcelessTransactionManager());
step.afterPropertiesSet();
return step;
}
public void testSimpleJob() throws Exception {
job.setSteps(new ArrayList());
AbstractStep step = new SimpleStep(getTasklet("foo", "bar"));
step.setJobRepository(repository);
step.setTransactionManager(new ResourcelessTransactionManager());
AbstractStep step = getStep("foo", "bar");
job.addStep(step);
step = new SimpleStep(getTasklet("spam"));
step.setJobRepository(repository);
step.setTransactionManager(new ResourcelessTransactionManager());
step = getStep("spam");
job.addStep(step);
JobInstance jobInstance = repository.createJobExecution(job, new JobParameters()).getJobInstance();
@@ -137,18 +132,19 @@ public class SimpleJobTests extends TestCase {
* is recovered ("skipped") on the second attempt (see retry policy
* definition above)...
*/
final ItemOrientedTasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
RepeatOperationsStep step = new RepeatOperationsStep();
step.setTasklet(module);
step.setChunkOperations(chunkOperations);
step.setJobRepository(repository);
step.setTransactionManager(new ResourcelessTransactionManager());
module.setItemWriter(new AbstractItemWriter() {
AbstractStep step = getStep(new String[] { "foo", "bar", "spam" });
// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
// RepeatOperationsStep step = new RepeatOperationsStep();
// step.setTasklet(module);
// step.setChunkOperations(chunkOperations);
step.setItemWriter(new AbstractItemWriter() {
public void write(Object data) throws Exception {
throw new RuntimeException("Try again Dummy!");
}
});
module.afterPropertiesSet();
step.afterPropertiesSet();
job.setSteps(Collections.singletonList(step));
JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());
@@ -162,17 +158,14 @@ public class SimpleJobTests extends TestCase {
}
public void testExceptionTerminates() throws Exception {
final ItemOrientedTasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
AbstractStep step = new SimpleStep(module);
step.setJobRepository(repository);
step.setTransactionManager(new ResourcelessTransactionManager());
module.setItemWriter(new AbstractItemWriter() {
// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
AbstractStep step = getStep(new String[] { "foo", "bar", "spam" });
step.setItemWriter(new AbstractItemWriter() {
public void write(Object data) throws Exception {
throw new RuntimeException("Foo");
}
});
module.afterPropertiesSet();
step.afterPropertiesSet();
job.setSteps(Collections.singletonList(step));
JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());

View File

@@ -18,6 +18,8 @@ package org.springframework.batch.execution.step.simple;
import junit.framework.TestCase;
import org.springframework.batch.core.tasklet.Tasklet;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.batch.item.writer.AbstractItemWriter;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.exception.handler.DefaultExceptionHandler;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
@@ -31,8 +33,7 @@ public class SimpleStepConfigurationTests extends TestCase {
SimpleStep configuration = new SimpleStep("foo");
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.SimpleStep#SimpleStepConfiguration()}.
* Test method for {@link org.springframework.batch.execution.step.simple.SimpleStep#SimpleStepConfiguration()}.
*/
public void testSimpleStepConfiguration() {
assertNotNull(configuration.getName());
@@ -43,23 +44,32 @@ public class SimpleStepConfigurationTests extends TestCase {
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.SimpleStep#SimpleStepConfiguration(org.springframework.batch.core.tasklet.Tasklet)}.
*
* @throws Exception
*/
public void testSimpleStepConfigurationTasklet() throws Exception {
Tasklet tasklet = new Tasklet() {
public ExitStatus execute() throws Exception {
return ExitStatus.FINISHED;
configuration = new SimpleStep();
configuration.setItemReader(new AbstractItemReader() {
public Object read() throws Exception {
// TODO Auto-generated method stub
return null;
}
};
configuration = new SimpleStep(tasklet);
});
configuration.setItemWriter(new AbstractItemWriter() {
public void write(Object item) throws Exception {
// TODO Auto-generated method stub
}
});
configuration.setJobRepository(new JobRepositorySupport());
configuration.setTransactionManager(new ResourcelessTransactionManager());
configuration.afterPropertiesSet();
}
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.SimpleStep#getCommitInterval()}.
* Test method for {@link org.springframework.batch.execution.step.simple.SimpleStep#getCommitInterval()}.
*/
public void testGetCommitInterval() {
assertEquals(1, configuration.getCommitInterval());
@@ -68,8 +78,7 @@ public class SimpleStepConfigurationTests extends TestCase {
}
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}.
* Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}.
*/
public void testGetExceptionHandler() {
assertNull(configuration.getExceptionHandler());
@@ -78,8 +87,7 @@ public class SimpleStepConfigurationTests extends TestCase {
}
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}.
* Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#getExceptionHandler()}.
*/
public void testSkipLimit() {
assertEquals(0, configuration.getSkipLimit());
@@ -88,8 +96,7 @@ public class SimpleStepConfigurationTests extends TestCase {
}
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.AbstractStep#getSkipLimit()}.
* Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#getSkipLimit()}.
*/
public void testGetSkipLimit() {
assertEquals(0, configuration.getSkipLimit());
@@ -98,8 +105,7 @@ public class SimpleStepConfigurationTests extends TestCase {
}
/**
* Test method for
* {@link org.springframework.batch.execution.step.simple.AbstractStep#isSaveExecutionAttributes()}.
* Test method for {@link org.springframework.batch.execution.step.simple.AbstractStep#isSaveExecutionAttributes()}.
*/
public void testIsSaveExecutionAttributes() {
assertEquals(false, configuration.isSaveExecutionAttributes());

View File

@@ -29,6 +29,7 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.JobSupport;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.domain.StepContribution;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
@@ -39,7 +40,6 @@ import org.springframework.batch.execution.repository.dao.MapJobDao;
import org.springframework.batch.execution.repository.dao.MapStepDao;
import org.springframework.batch.execution.scope.StepScope;
import org.springframework.batch.execution.scope.StepSynchronizationManager;
import org.springframework.batch.execution.tasklet.ItemOrientedTasklet;
import org.springframework.batch.item.ExecutionAttributes;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -84,31 +84,21 @@ public class SimpleStepExecutorTests extends TestCase {
return new ListItemReader(Arrays.asList(args));
}
/**
* @param strings
* @return
* @throws Exception
*/
private Tasklet getTasklet(String[] strings) throws Exception {
ItemOrientedTasklet module = new ItemOrientedTasklet();
module.setItemWriter(processor);
module.setItemReader(getReader(strings));
module.afterPropertiesSet();
return module;
private AbstractStep getStep(String[] strings) throws Exception {
SimpleStep step = new SimpleStep();
step.setItemWriter(processor);
step.setItemReader(getReader(strings));
step.setJobRepository(new JobRepositorySupport());
step.setTransactionManager(transactionManager);
step.afterPropertiesSet();
return step;
}
/*
* (non-Javadoc)
*
* @see junit.framework.TestCase#setUp()
*/
protected void setUp() throws Exception {
super.setUp();
transactionManager = new ResourcelessTransactionManager();
stepConfiguration = new SimpleStep();
stepConfiguration.setTasklet(getTasklet(new String[] { "foo", "bar", "spam" }));
stepConfiguration.setJobRepository(new JobRepositorySupport());
stepConfiguration.setTransactionManager(transactionManager);
stepConfiguration = getStep(new String[] { "foo", "bar", "spam" });
stepExecutor = (SimpleStepExecutor) stepConfiguration.createStepExecutor();
template = new RepeatTemplate();

View File

@@ -18,7 +18,8 @@
class="org.springframework.batch.execution.configuration.MapJobRegistry" />
<bean id="simpleJob"
class="org.springframework.batch.execution.job.simple.SimpleJob" abstract="true">
class="org.springframework.batch.execution.job.simple.SimpleJob"
abstract="true">
<property name="restartable" value="true" />
<property name="jobRepository" ref="simpleJobRepository" />
</bean>
@@ -35,8 +36,8 @@
<property name="useParent" value="true" />
</bean>
</property>
<property name="jobRepository" ref="jobRegistry"/>
<property name="transactionManager" ref="transactionManager"/>
<property name="jobRepository" ref="jobRegistry" />
<property name="transactionManager" ref="transactionManager" />
</bean>
<bean id="transactionManager"