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

OPEN - issue BATCH-404: FactoryBeans for step configuration

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

SPlit item processor implementations
This commit is contained in:
dsyer
2008-03-03 17:50:16 +00:00
parent 7ccae72cc3
commit 36801d7d5d
4 changed files with 83 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java
View File

@@ -52,6 +52,8 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
private ItemProcessor itemProcessor;
private RepeatTemplate stepOperations;
/**
* Set the commit interval.
*
@@ -96,6 +98,15 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
this.listeners = listeners;
}
/**
* Protected getter for the step operations to make them available in
* subclasses.
* @return the step operations
*/
protected RepeatTemplate getStepOperations() {
return stepOperations;
}
/**
* Public setter for the {@link TaskExecutor}. If this is set, then it will
* be used to execute the chunk processing inside the {@link Step}.
@@ -172,7 +183,7 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
StepListener[] stepListeners = helper.getStepListeners(listeners);
itemReader = helper.getItemReader(itemReader, listeners);
itemWriter = helper.getItemWriter(itemWriter, listeners);
RepeatTemplate stepOperations = new RepeatTemplate();
stepOperations = new RepeatTemplate();
stepOperations = (RepeatTemplate) helper.getStepOperations(stepOperations, listeners);
// In case they are used by subclasses:
@@ -185,10 +196,11 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setTaskExecutor(taskExecutor);
stepOperations = repeatTemplate;
step.setStepOperations(stepOperations);
}
KitchenSinkItemProcessor itemProcessor = new KitchenSinkItemProcessor(itemReader, itemWriter);
step.setStepOperations(stepOperations);
ItemSkipPolicyItemProcessor itemProcessor = new ItemSkipPolicyItemProcessor(itemReader, itemWriter);
if (skipLimit > 0) {
/*
@@ -209,4 +221,5 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
step.setItemProcessor(itemProcessor);
}
}

56
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/KitchenSinkItemProcessor.java → spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemSkipPolicyItemProcessor.java
View File

@@ -19,23 +19,14 @@ import org.springframework.batch.core.domain.ItemSkipPolicy;
import org.springframework.batch.core.domain.StepContribution;
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.repeat.ExitStatus;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryOperations;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.support.RetryTemplate;
/**
* @author Dave Syer
*
*/
public class KitchenSinkItemProcessor extends SimpleItemProcessor {
private RetryOperations retryOperations = new RetryTemplate();
private ItemReaderRetryCallback retryCallback;
public class ItemSkipPolicyItemProcessor extends SimpleItemProcessor {
private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
@@ -43,7 +34,7 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor {
* @param itemReader
* @param itemWriter
*/
public KitchenSinkItemProcessor(ItemReader itemReader, ItemWriter itemWriter) {
public ItemSkipPolicyItemProcessor(ItemReader itemReader, ItemWriter itemWriter) {
super(itemReader, itemWriter);
}
@@ -54,37 +45,13 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor {
this.itemSkipPolicy = itemSkipPolicy;
}
/**
* Public setter for the {@link RetryOperations}.
* @param retryOperations the {@link RetryOperations} to set
*/
public void setRetryOperations(RetryOperations retryOperations) {
this.retryOperations = retryOperations;
}
/**
* Public setter for the ItemReaderRetryCallback. TODO: get rid of this.
* @param retryCallback the retryCallback to set
*/
public void setRetryCallback(ItemReaderRetryCallback retryCallback) {
this.retryCallback = retryCallback;
}
/**
* Execute the business logic, delegating to the reader and writer.
* Subclasses could extend the behaviour as long as they always return the
* value of this method call in their superclass.<br/>
*
* Read from the {@link ItemReader} and process (if not null) with the
* {@link ItemWriter}. If a {@link RetryCallback} is provided, then the
* call to {@link ItemWriter} is wrapped in a stateful retry. In that case
* 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).<br/>
* {@link ItemWriter}.<br/>
*
* If there is an exception and the reader or writer implements
* {@link Skippable} then the skip method is called.
@@ -96,10 +63,6 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor {
public ExitStatus process(StepContribution contribution) throws Exception {
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
if (retryCallback != null) {
return new ExitStatus(retryOperations.execute(retryCallback) != null);
}
try {
exitStatus = super.process(contribution);
@@ -107,7 +70,7 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor {
}
catch (Exception e) {
if (retryCallback == null && itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) {
if (itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) {
contribution.incrementSkipCount();
skip();
}
@@ -122,19 +85,12 @@ public class KitchenSinkItemProcessor extends SimpleItemProcessor {
}
/**
* 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.
* Mark the current item as skipped if possible. 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 (getItemReader() instanceof Skippable) {
((Skippable) getItemReader()).skip();
}

71
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java
View File

@@ -16,10 +16,15 @@
package org.springframework.batch.execution.step.support;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.domain.StepContribution;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler;
import org.springframework.batch.retry.RetryOperations;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
@@ -74,25 +79,25 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
*/
protected void applyConfiguration(ItemOrientedStep step) {
// Ensure exception handler always rethrows. N.B. no skips ever actually
// take place.
if (retryPolicy != null) {
// TODO: actually we need to co-ordinate the retry policy with the
// exception handler limit, so this is a hack for now.
super.setSkipLimit(Integer.MAX_VALUE);
}
super.applyConfiguration(step);
if (retryPolicy != null) {
// TODO: actually we need to co-ordinate the retry policy with the
// exception handler limit, so this is a hack for now.
getStepOperations().setExceptionHandler(new SimpleLimitExceptionHandler(Integer.MAX_VALUE));
ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), getKeyGenerator(),
getItemWriter());
ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy);
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(itemProviderRetryPolicy);
KitchenSinkItemProcessor itemProcessor = (KitchenSinkItemProcessor) getItemProcessor();
itemProcessor.setRetryOperations(retryTemplate);
itemProcessor.setRetryCallback(retryCallback);
StatefulRetryItemProcessor itemProcessor = new StatefulRetryItemProcessor(getItemReader(), getItemWriter(), retryTemplate, retryCallback);
step.setItemProcessor(itemProcessor);
}
}
@@ -115,4 +120,48 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
}
private static class StatefulRetryItemProcessor extends SimpleItemProcessor {
final private RetryOperations retryOperations;
final private ItemReaderRetryCallback retryCallback;
/**
* @param itemReader
* @param itemWriter
* @param retryCallback
* @param retryTemplate
*/
public StatefulRetryItemProcessor(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, ItemReaderRetryCallback retryCallback) {
super(itemReader, itemWriter);
this.retryOperations = retryTemplate;
this.retryCallback = retryCallback;
}
/**
* Execute the business logic, delegating to the reader and writer.
* Subclasses could extend the behaviour as long as they always return
* the value of this method call in their superclass.<br/>
*
* 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. In that case 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).<br/>
*
* @param contribution the current step
* @return {@link ExitStatus#CONTINUABLE} if there is more processing to
* do
* @throws Exception if there is an error
*/
public ExitStatus process(StepContribution contribution) throws Exception {
return new ExitStatus(retryOperations.execute(retryCallback) != null);
}
}
}