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

Merge pull request #11 from magott/master

Browse Source 
BATCH-1783: Forcing step termination when ChunkListener fails
This commit is contained in:
Dave Syer
2011-10-19 01:26:53 -07:00
parent 6832201363 af84be4f75
commit b075851184
3 changed files with 151 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
View File

@@ -26,8 +26,10 @@ import java.util.Map;
import org.springframework.batch.classify.BinaryExceptionClassifier;
import org.springframework.batch.classify.Classifier;
import org.springframework.batch.classify.SubclassClassifier;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.step.FatalStepExecutionException;
import org.springframework.batch.core.step.skip.CompositeSkipPolicy;
import org.springframework.batch.core.step.skip.ExceptionClassifierSkipPolicy;
@@ -78,6 +80,7 @@ import org.springframework.util.Assert;
*
* @author Dave Syer
* @author Robert Kasanicky
* @author Morten Andersen-Gott
*
*/
public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S> {
@@ -120,7 +123,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* {@link #setIsReaderTransactionalQueue(boolean) transactional queue flag}
* being false (the default).
*
* @param keyGenerator the {@link KeyGenerator} to set
* @param keyGenerator
* the {@link KeyGenerator} to set
*/
public void setKeyGenerator(KeyGenerator keyGenerator) {
this.keyGenerator = keyGenerator;
@@ -131,7 +135,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* properties are ignored (retryLimit, backOffPolicy,
* retryableExceptionClasses).
*
* @param retryPolicy a stateless {@link RetryPolicy}
* @param retryPolicy
* a stateless {@link RetryPolicy}
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
@@ -142,7 +147,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* limit. Note this limit includes the initial attempt to process the item,
* therefore <code>retryLimit == 1</code> by default.
*
* @param retryLimit the retry limit to set, must be greater or equal to 1.
* @param retryLimit
* the retry limit to set, must be greater or equal to 1.
*/
public void setRetryLimit(int retryLimit) {
this.retryLimit = retryLimit;
@@ -162,8 +168,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* This property is ignored if the
* {@link #setRetryContextCache(RetryContextCache)} is set directly.
*
* @param cacheCapacity the cache capacity to set (greater than 0 else
* ignored)
* @param cacheCapacity
* the cache capacity to set (greater than 0 else ignored)
*/
public void setCacheCapacity(int cacheCapacity) {
this.cacheCapacity = cacheCapacity;
@@ -173,7 +179,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* Override the default retry context cache for retry of chunk processing.
* If this property is set then {@link #setCacheCapacity(int)} is ignored.
*
* @param retryContextCache the {@link RetryContextCache} to set
* @param retryContextCache
* the {@link RetryContextCache} to set
*/
public void setRetryContextCache(RetryContextCache retryContextCache) {
this.retryContextCache = retryContextCache;
@@ -183,7 +190,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* Public setter for the retryable exceptions classifier map (from throwable
* class to boolean, true is retryable).
*
* @param retryableExceptionClasses the retryableExceptionClasses to set
* @param retryableExceptionClasses
* the retryableExceptionClasses to set
*/
public void setRetryableExceptionClasses(Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses) {
this.retryableExceptionClasses = retryableExceptionClasses;
@@ -192,7 +200,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
/**
* Public setter for the {@link BackOffPolicy}.
*
* @param backOffPolicy the {@link BackOffPolicy} to set
* @param backOffPolicy
* the {@link BackOffPolicy} to set
*/
public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
this.backOffPolicy = backOffPolicy;
@@ -201,7 +210,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
/**
* Public setter for the {@link RetryListener}s.
*
* @param retryListeners the {@link RetryListener}s to set
* @param retryListeners
* the {@link RetryListener}s to set
*/
public void setRetryListeners(RetryListener... retryListeners) {
this.retryListeners = retryListeners;
@@ -213,7 +223,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* exception propagated until the limit is reached. If it is zero then all
* exceptions will be propagated from the chunk and cause the step to abort.
*
* @param skipLimit the value to set. Default is 0 (never skip).
* @param skipLimit
* the value to set. Default is 0 (never skip).
*/
public void setSkipLimit(int skipLimit) {
this.skipLimit = skipLimit;
@@ -225,7 +236,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* The {@link #setSkippableExceptionClasses(Map) skippableExceptionClasses}
* are also ignored if this is set.
*
* @param skipPolicy the {@link SkipPolicy} to set
* @param skipPolicy
* the {@link SkipPolicy} to set
*/
public void setSkipPolicy(SkipPolicy skipPolicy) {
this.skipPolicy = skipPolicy;
@@ -239,7 +251,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* <p/>
* Defaults to all no exception.
*
* @param exceptionClasses defaults to <code>Exception</code>
* @param exceptionClasses
* defaults to <code>Exception</code>
*/
public void setSkippableExceptionClasses(Map<Class<? extends Throwable>, Boolean> exceptionClasses) {
this.skippableExceptionClasses = exceptionClasses;
@@ -254,7 +267,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* <p/>
* Defaults is empty.
*
* @param noRollbackExceptionClasses the exception classes to set
* @param noRollbackExceptionClasses
* the exception classes to set
*/
public void setNoRollbackExceptionClasses(Collection<Class<? extends Throwable>> noRollbackExceptionClasses) {
this.noRollbackExceptionClasses = noRollbackExceptionClasses;
@@ -272,7 +286,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* the provided transaction attributes.
*
* @return an exception classifier: maps to true if an exception should
* cause rollback
* cause rollback
*/
protected Classifier<Throwable, Boolean> getRollbackClassifier() {
@@ -328,11 +342,11 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
@SuppressWarnings("unchecked")
protected void applyConfiguration(TaskletStep step) {
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class,
JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
TransactionException.class, FatalStepExecutionException.class, SkipListenerFailedException.class,
SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class,
Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, TransactionException.class,
FatalStepExecutionException.class, SkipListenerFailedException.class, SkipPolicyFailedException.class,
RetryException.class, JobInterruptedException.class, Error.class);
super.applyConfiguration(step);
}
@@ -347,8 +361,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
if (stream instanceof ItemReader<?>) {
streamIsReader = true;
chunkMonitor.registerItemStream(stream);
}
else {
} else {
step.registerStream(stream);
}
}
@@ -361,11 +374,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
if (!concurrent) {
chunkMonitor.setItemReader(itemReader);
}
else {
} else {
logger.warn("Asynchronous TaskExecutor detected (" + taskExecutor.getClass()
+ ") with ItemStream reader. This is probably an error, "
+ "and may lead to incorrect restart data being stored.");
+ ") with ItemStream reader. This is probably an error, " + "and may lead to incorrect restart data being stored.");
}
}
}
@@ -378,10 +389,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
SkipPolicy readSkipPolicy = createSkipPolicy();
readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy);
FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(),
getChunkOperations());
chunkProvider.setMaxSkipsOnRead(Math.max(getCommitInterval(),
FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ));
FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(), getChunkOperations());
chunkProvider.setMaxSkipsOnRead(Math.max(getCommitInterval(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ));
chunkProvider.setSkipPolicy(readSkipPolicy);
chunkProvider.setRollbackClassifier(getRollbackClassifier());
@@ -394,16 +403,14 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
*/
protected SkipPolicy createSkipPolicy() {
SkipPolicy skipPolicy = this.skipPolicy;
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(
skippableExceptionClasses);
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(skippableExceptionClasses);
map.put(ForceRollbackForWriteSkipException.class, true);
LimitCheckingItemSkipPolicy limitCheckingItemSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, map);
if (skipPolicy == null) {
Assert.state(!(skippableExceptionClasses.isEmpty() && skipLimit > 0),
"If a skip limit is provided then skippable exceptions must also be specified");
skipPolicy = limitCheckingItemSkipPolicy;
}
else if (limitCheckingItemSkipPolicy != null) {
} else if (limitCheckingItemSkipPolicy != null) {
skipPolicy = new CompositeSkipPolicy(new SkipPolicy[] { skipPolicy, limitCheckingItemSkipPolicy });
}
return skipPolicy;
@@ -417,8 +424,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
BatchRetryTemplate batchRetryTemplate = configureRetry();
FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(getItemProcessor(),
getItemWriter(), batchRetryTemplate);
FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(getItemProcessor(), getItemWriter(),
batchRetryTemplate);
chunkProcessor.setBuffering(!isReaderTransactionalQueue());
chunkProcessor.setProcessorTransactional(processorTransactional);
@@ -442,8 +449,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
RetryPolicy retryPolicy = this.retryPolicy;
SimpleRetryPolicy simpleRetryPolicy = null;
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(
retryableExceptionClasses);
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>(retryableExceptionClasses);
map.put(ForceRollbackForWriteSkipException.class, true);
simpleRetryPolicy = new SimpleRetryPolicy(retryLimit, map);
@@ -451,8 +457,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
Assert.state(!(retryableExceptionClasses.isEmpty() && retryLimit > 0),
"If a retry limit is provided then retryable exceptions must also be specified");
retryPolicy = simpleRetryPolicy;
}
else if ((!retryableExceptionClasses.isEmpty() && retryLimit > 0)) {
} else if ((!retryableExceptionClasses.isEmpty() && retryLimit > 0)) {
CompositeRetryPolicy compositeRetryPolicy = new CompositeRetryPolicy();
compositeRetryPolicy.setPolicies(new RetryPolicy[] { retryPolicy, simpleRetryPolicy });
retryPolicy = compositeRetryPolicy;
@@ -469,8 +474,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
// Co-ordinate the retry policy with the exception handler:
RepeatOperations stepOperations = getStepOperations();
if (stepOperations instanceof RepeatTemplate) {
SimpleRetryExceptionHandler exceptionHandler = new SimpleRetryExceptionHandler(retryPolicyWrapper,
getExceptionHandler(), nonRetryableExceptionClasses);
SimpleRetryExceptionHandler exceptionHandler = new SimpleRetryExceptionHandler(retryPolicyWrapper, getExceptionHandler(),
nonRetryableExceptionClasses);
((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler);
}
@@ -478,8 +483,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
if (cacheCapacity > 0) {
batchRetryTemplate.setRetryContextCache(new MapRetryContextCache(cacheCapacity));
}
}
else {
} else {
batchRetryTemplate.setRetryContextCache(retryContextCache);
}
@@ -501,8 +505,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
map.put(fatal, neverRetryPolicy);
}
SubclassClassifier<Throwable, RetryPolicy> classifier = new SubclassClassifier<Throwable, RetryPolicy>(
retryPolicy);
SubclassClassifier<Throwable, RetryPolicy> classifier = new SubclassClassifier<Throwable, RetryPolicy>(retryPolicy);
classifier.setTypeMap(map);
ExceptionClassifierRetryPolicy retryPolicyWrapper = new ExceptionClassifierRetryPolicy();
@@ -515,7 +518,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* Wrap a {@link SkipPolicy} and make it consistent with known fatal
* exceptions.
*
* @param skipPolicy an existing skip policy
* @param skipPolicy
* an existing skip policy
* @return a skip policy that will not skip fatal exceptions
*/
private SkipPolicy getFatalExceptionAwareProxy(SkipPolicy skipPolicy) {
@@ -560,4 +564,44 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
nonRetryableExceptionClasses = (List<Class<? extends Throwable>>) exceptions;
}
@Override
protected void registerChunkListeners(TaskletStep step, StepListener listener) {
super.registerChunkListeners(step, new TerminateOnExceptionChunkListenerDelegate((ChunkListener) listener));
}
/**
* ChunkListener that wraps exceptions thrown from the ChunkListener in
* {@link FatalStepExecutionException} to force termination of StepExecution
*
* ChunkListeners shoulnd't throw exceptions and expect continued
* processing, they must be handled in the implementation or the step will
* terminate
*
*/
private class TerminateOnExceptionChunkListenerDelegate implements ChunkListener {
private ChunkListener chunkListener;
TerminateOnExceptionChunkListenerDelegate(ChunkListener chunkListener) {
this.chunkListener = chunkListener;
}
public void beforeChunk() {
try {
chunkListener.beforeChunk();
} catch (Throwable t) {
throw new FatalStepExecutionException("ChunkListener threw exception, rethrowing as fatal", t);
}
}
public void afterChunk() {
try {
chunkListener.afterChunk();
} catch (Throwable t) {
throw new FatalStepExecutionException("ChunkListener threw exception, rethrowing as fatal", t);
}
}
}
}

12
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -590,15 +590,21 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
step.registerStepExecutionListener((StepExecutionListener) listener);
}
if (listener instanceof ChunkListener) {
step.registerChunkListener((ChunkListener) listener);
registerChunkListeners(step, listener);
}
}
}
step.setStepExecutionListeners(BatchListenerFactoryHelper.getListeners(listeners, StepExecutionListener.class)
.toArray(new StepExecutionListener[] {}));
step.setChunkListeners(BatchListenerFactoryHelper.getListeners(listeners, ChunkListener.class).toArray(
new ChunkListener[] {}));
for(ChunkListener chunkListener: BatchListenerFactoryHelper.getListeners(listeners, ChunkListener.class)){
registerChunkListeners(step,chunkListener);
}
}
protected void registerChunkListeners(TaskletStep step, StepListener listener) {
step.registerChunkListener((ChunkListener) listener);
}
/**

52
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
View File

@@ -1,8 +1,11 @@
package org.springframework.batch.core.step.item;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.batch.core.BatchStatus.FAILED;
import java.util.ArrayList;
import java.util.Arrays;
@@ -17,12 +20,15 @@ import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.FatalStepExecutionException;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.transaction.interceptor.RollbackRuleAttribute;
@@ -90,6 +96,32 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
}
@Test
public void testBeforeChunkListenerException() throws Exception{
factory.setListeners(new StepListener []{new ExceptionThrowingChunkListener(true)});
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(FAILED, stepExecution.getStatus());
assertEquals(FAILED.toString(), stepExecution.getExitStatus().getExitCode());
assertTrue(stepExecution.getCommitCount() == 0);//Make sure exception was thrown in after, not before
Throwable e = stepExecution.getFailureExceptions().get(0);
assertThat(e, instanceOf(FatalStepExecutionException.class));
assertThat(e.getCause(), instanceOf(IllegalArgumentException.class));
}
@Test
public void testAfterChunkListenerException() throws Exception{
factory.setListeners(new StepListener []{new ExceptionThrowingChunkListener(false)});
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(FAILED, stepExecution.getStatus());
assertEquals(FAILED.toString(), stepExecution.getExitStatus().getExitCode());
assertTrue(stepExecution.getCommitCount() > 0);//Make sure exception was thrown in after, not before
Throwable e = stepExecution.getFailureExceptions().get(0);
assertThat(e, instanceOf(FatalStepExecutionException.class));
assertThat(e.getCause(), instanceOf(IllegalArgumentException.class));
}
@Test
public void testOverrideWithoutChangingRollbackRules() throws Exception {
@@ -525,5 +557,25 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
}
return map;
}
class ExceptionThrowingChunkListener implements ChunkListener{
private boolean throwBefore = true;
public ExceptionThrowingChunkListener(boolean throwBefore) {
this.throwBefore = throwBefore;
}
public void beforeChunk() {
if(throwBefore){
throw new IllegalArgumentException("Planned exception");
}
}
public void afterChunk() {
throw new IllegalArgumentException("Planned exception");
}
}
}