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

BATCH-873: add skipPolicy to FultTolerantStepFactoryBean

Browse Source
This commit is contained in:
dsyer
2009-12-31 10:10:42 +00:00
parent 5b348c803a
commit 1f6c4cce79
5 changed files with 139 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -28,7 +28,9 @@ import org.springframework.batch.classify.Classifier;
import org.springframework.batch.classify.SubclassClassifier;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.skip.ExceptionClassifierSkipPolicy;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
import org.springframework.batch.core.step.skip.NonSkippableReadException;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
@@ -76,6 +78,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
private Map<Class<? extends Throwable>, Boolean> skippableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
private Collection<Class<? extends Throwable>> nonSkippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Collection<Class<? extends Throwable>> noRollbackExceptionClasses = new HashSet<Class<? extends Throwable>>();
private Map<Class<? extends Throwable>, Boolean> retryableExceptionClasses = new HashMap<Class<? extends Throwable>, Boolean>();
@@ -88,6 +92,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
private int skipLimit = 0;
private SkipPolicy skipPolicy;
private BackOffPolicy backOffPolicy;
private RetryListener[] retryListeners;
@@ -207,6 +213,18 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
this.skipLimit = skipLimit;
}
/**
* A {@link SkipPolicy} that determines the outcome of an exception when
* processing an item. Overrides the {@link #setSkipLimit(int) skipLimit}.
* The {@link #setSkippableExceptionClasses(Map) skippableExceptionClasses}
* are also ignored if this is set.
*
* @param skipPolicy the {@link SkipPolicy} to set
*/
public void setSkipPolicy(SkipPolicy skipPolicy) {
this.skipPolicy = skipPolicy;
}
/**
* Exception classes that when raised won't crash the job but will result in
* the item which handling caused the exception being skipped. Any exception
@@ -300,14 +318,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
}
@SuppressWarnings("unchecked")
@Override
protected void applyConfiguration(TaskletStep step) {
addFatalExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
super.applyConfiguration(step);
}
@@ -351,7 +367,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
@Override
protected SimpleChunkProvider<T> configureChunkProvider() {
SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses());
SkipPolicy readSkipPolicy = skipPolicy != null ? skipPolicy : new LimitCheckingItemSkipPolicy(skipLimit,
getSkippableExceptionClasses());
readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy);
FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(),
getChunkOperations());
chunkProvider.setSkipPolicy(readSkipPolicy);
@@ -374,7 +392,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
chunkProcessor.setBuffering(!isReaderTransactionalQueue());
chunkProcessor.setProcessorTransactional(processorTransactional);
SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses());
SkipPolicy writeSkipPolicy = skipPolicy != null ? skipPolicy : new LimitCheckingItemSkipPolicy(skipLimit,
getSkippableExceptionClasses());
writeSkipPolicy = getFatalExceptionAwareProxy(writeSkipPolicy);
chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
chunkProcessor.setRollbackClassifier(getRollbackClassifier());
@@ -409,7 +429,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
retryPolicy = new SimpleRetryPolicy(retryLimit, map);
}
RetryPolicy retryPolicyWrapper = fatalExceptionAwareProxy(retryPolicy);
RetryPolicy retryPolicyWrapper = getFatalExceptionAwareProxy(retryPolicy);
BatchRetryTemplate batchRetryTemplate = new BatchRetryTemplate();
if (backOffPolicy != null) {
@@ -444,7 +464,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never
* retries explicitly non-retryable exceptions.
*/
private RetryPolicy fatalExceptionAwareProxy(final RetryPolicy retryPolicy) {
private RetryPolicy getFatalExceptionAwareProxy(RetryPolicy retryPolicy) {
NeverRetryPolicy neverRetryPolicy = new NeverRetryPolicy();
Map<Class<? extends Throwable>, RetryPolicy> map = new HashMap<Class<? extends Throwable>, RetryPolicy>();
@@ -462,15 +482,41 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
}
private void addFatalExceptionIfMissing(Class<? extends Throwable>... classes) {
Map<Class<? extends Throwable>, Boolean> exceptions = new HashMap<Class<? extends Throwable>, Boolean>(
skippableExceptionClasses);
for (Class<? extends Throwable> cls : classes) {
if (!exceptions.containsKey(cls)) {
exceptions.put(cls, false);
/**
* Wrap a {@link SkipPolicy} and make it consistent with known fatal
* exceptions.
*
* @param skipPolicy an existing skip policy
* @return a skip policy that will not skip fatal exceptions
*/
private SkipPolicy getFatalExceptionAwareProxy(SkipPolicy skipPolicy) {
NeverSkipItemSkipPolicy neverSkipPolicy = new NeverSkipItemSkipPolicy();
Map<Class<? extends Throwable>, SkipPolicy> map = new HashMap<Class<? extends Throwable>, SkipPolicy>();
for (Class<? extends Throwable> fatal : nonSkippableExceptionClasses) {
map.put(fatal, neverSkipPolicy);
}
SubclassClassifier<Throwable, SkipPolicy> classifier = new SubclassClassifier<Throwable, SkipPolicy>(skipPolicy);
classifier.setTypeMap(map);
ExceptionClassifierSkipPolicy skipPolicyWrapper = new ExceptionClassifierSkipPolicy();
skipPolicyWrapper.setExceptionClassifier(classifier);
return skipPolicyWrapper;
}
@SuppressWarnings("unchecked")
private void addNonSkippableExceptionIfMissing(Class... cls) {
List exceptions = new ArrayList<Class<? extends Throwable>>();
for (Class exceptionClass : nonSkippableExceptionClasses) {
exceptions.add(exceptionClass);
}
for (Class fatal : cls) {
if (!exceptions.contains(fatal)) {
exceptions.add(fatal);
}
}
skippableExceptionClasses = exceptions;
nonSkippableExceptionClasses = exceptions;
}
@SuppressWarnings("unchecked")

54
spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/ExceptionClassifierSkipPolicy.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2006-2010 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.core.step.skip;
import org.springframework.batch.classify.SubclassClassifier;
/**
* A {@link SkipPolicy} that depends on an exception classifier to make its
* decision, and then delegates to the classifier result.
*
* @author Dave Syer
*
* @see SubclassClassifier
*/
public class ExceptionClassifierSkipPolicy implements SkipPolicy {
private SubclassClassifier<Throwable, SkipPolicy> classifier;
/**
* The classifier that will be used to choose a delegate policy.
*
* @param classifier the classifier to use to choose a delegate policy
*/
public void setExceptionClassifier(SubclassClassifier<Throwable, SkipPolicy> classifier) {
this.classifier = classifier;
}
/**
* Consult the classifier and find a delegate policy, and then use that to
* determine the outcome.
*
* @param t the throwable to consider
* @param skipCount the current skip count
* @return true if the exception can be skipped
* @throws SkipLimitExceededException if a limit is exceeded
*/
public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
return classifier.classify(t).shouldSkip(t, skipCount);
}
}

13
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
View File

@@ -26,8 +26,10 @@ import java.util.Collection;
import java.util.Map;
import org.junit.Test;
import org.springframework.batch.classify.SubclassClassifier;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.item.SimpleChunkProcessor;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.support.CompositeItemStream;
@@ -103,7 +105,7 @@ public class ChunkElementParserTests {
@Test
public void testInheritSkippable() throws Exception {
Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s1", getContext());
assertEquals(11, skippable.size());
assertEquals(5, skippable.size());
containsClassified(skippable, NullPointerException.class, true);
containsClassified(skippable, ArithmeticException.class, true);
containsClassified(skippable, CannotAcquireLockException.class, false);
@@ -113,7 +115,7 @@ public class ChunkElementParserTests {
@Test
public void testInheritSkippableWithNoMerge() throws Exception {
Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s2", getContext());
assertEquals(9, skippable.size());
assertEquals(3, skippable.size());
containsClassified(skippable, NullPointerException.class, true);
assertFalse(skippable.containsKey(ArithmeticException.class));
containsClassified(skippable, CannotAcquireLockException.class, false);
@@ -194,8 +196,11 @@ public class ChunkElementParserTests {
Object tasklet = ReflectionTestUtils.getField(step, "tasklet");
Object chunkProvider = ReflectionTestUtils.getField(tasklet, "chunkProvider");
Object skipPolicy = ReflectionTestUtils.getField(chunkProvider, "skipPolicy");
Object classifier = ReflectionTestUtils.getField(skipPolicy, "skippableExceptionClassifier");
return (Map<Class<? extends Throwable>, Boolean>) ReflectionTestUtils.getField(classifier, "classified");
SubclassClassifier<Throwable, SkipPolicy> classifier = (SubclassClassifier<Throwable, SkipPolicy>) ReflectionTestUtils
.getField(skipPolicy, "classifier");
Object limitPolicy = classifier.classify(new Exception());
Object limitClassifier = ReflectionTestUtils.getField(limitPolicy, "skippableExceptionClassifier");
return (Map<Class<? extends Throwable>, Boolean>) ReflectionTestUtils.getField(limitClassifier, "classified");
}
@SuppressWarnings("unchecked")

16
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -6,6 +6,7 @@ import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -32,6 +33,7 @@ import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.SkipListenerSupport;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
@@ -188,6 +190,18 @@ public class FaultTolerantStepFactoryBeanTests {
.getName()));
}
/**
* Check items causing errors are skipped as expected.
*/
@Test
public void testReadSkipWithPolicy() throws Exception {
// Should be ignored
factory.setSkipLimit(0);
factory.setSkipPolicy(new LimitCheckingItemSkipPolicy(2, Collections
.<Class<? extends Throwable>, Boolean> singletonMap(Exception.class, true)));
testReadSkip();
}
/**
* Check to make sure that ItemStreamException can be skipped. (see
* BATCH-915)
@@ -426,7 +440,7 @@ public class FaultTolerantStepFactoryBeanTests {
step.execute(stepExecution);
// 1,3 skipped inside a committed chunk. 5 tripped the skip
// limit but it was skipped in a chunk that rolled back, so
// limit but it was skipped in a chunk that rolled back, so
// it will re-appear on a restart and the listener is not called.
assertEquals(2, listenerCalls.size());
assertEquals(2, stepExecution.getReadSkipCount());

2
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml
View File

@@ -45,7 +45,7 @@
<step id="standalone2" parent="baseStep">
<tasklet>
<chunk reader="reader" writer="writer" commit-interval="5" />
<chunk reader="reader" writer="writer" commit-interval="5"/>
<transaction-attributes propagation="REQUIRED" isolation="DEFAULT" />
</tasklet>
</step>