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

BATCH-1529: added new exception type for skip policy errors

Browse Source
This commit is contained in:
dsyer
2010-04-05 07:57:49 +00:00
parent 1287b2f53e
commit 6473473d78
6 changed files with 171 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -236,7 +236,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
// allows us to continue
throw e;
}
else if (itemProcessSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
else if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
// If we are not re-throwing then we should check if
// this is skippable
contribution.incrementProcessSkipCount();
@@ -268,7 +268,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
public O recover(RetryContext context) throws Exception {
Throwable e = context.getLastThrowable();
if (itemProcessSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
contribution.incrementProcessSkipCount();
iterator.remove(e);
logger.debug("Skipping after failed process", e);
@@ -376,7 +376,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
* do any scanning. We can just bomb out with a retry
* exhausted.
*/
if (!itemWriteSkipPolicy.shouldSkip(context.getLastThrowable(), -1)) {
if (!shouldSkip(itemWriteSkipPolicy, context.getLastThrowable(), -1)) {
throw new ExhaustedRetryException(
"Retry exhausted after last attempt in recovery path, but exception is not skippable.",
context.getLastThrowable());
@@ -442,6 +442,23 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
}
}
/**
* Convenience method for calling process skip policy, so that it can be
* called from multiple places.
*
* @param policy the skip policy
* @param e the cause of the skip
* @param skipCount the current skip count
*/
private boolean shouldSkip(SkipPolicy policy, Throwable e, int skipCount) {
try {
return policy.shouldSkip(e, skipCount);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e);
}
}
private Object getInputKey(I item) {
if (keyGenerator == null) {
return item;
@@ -463,7 +480,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
private void checkSkipPolicy(Chunk<I>.ChunkIterator inputIterator, Chunk<O>.ChunkIterator outputIterator,
Throwable e, StepContribution contribution) {
logger.debug("Checking skip policy after failed write");
if (itemWriteSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
if (shouldSkip(itemWriteSkipPolicy, e, contribution.getStepSkipCount())) {
contribution.incrementWriteSkipCount();
inputIterator.remove();
outputIterator.remove(e);
@@ -498,7 +515,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
outputIterator.remove();
}
catch (Exception e) {
if (!itemWriteSkipPolicy.shouldSkip(e, -1) && !rollbackClassifier.classify(e)) {
if (!shouldSkip(itemWriteSkipPolicy, e, -1) && !rollbackClassifier.classify(e)) {
inputIterator.remove();
outputIterator.remove();
}

18
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java
View File

@@ -69,7 +69,7 @@ public class FaultTolerantChunkProvider<I> extends SimpleChunkProvider<I> {
}
catch (Exception e) {
if (skipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementReadSkipCount();
chunk.skip(e);
@@ -99,4 +99,20 @@ public class FaultTolerantChunkProvider<I> extends SimpleChunkProvider<I> {
}
}
/**
* Convenience method for calling process skip policy.
*
* @param policy the skip policy
* @param e the cause of the skip
* @param skipCount the current skip count
*/
private boolean shouldSkip(SkipPolicy policy, Throwable e, int skipCount) {
try {
return policy.shouldSkip(e, skipCount);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e);
}
}
}

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

@@ -35,6 +35,7 @@ import org.springframework.batch.core.step.skip.NonSkippableReadException;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.core.step.skip.SkipPolicyFailedException;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
@@ -321,9 +322,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
@Override
protected void applyConfiguration(TaskletStep step) {
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
super.applyConfiguration(step);
}

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

@@ -0,0 +1,39 @@
/*
* 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.core.step.skip;
import org.springframework.batch.core.UnexpectedJobExecutionException;
/**
* Special exception to indicate a failure in a skip policy. These need
* special treatment in the framework in case a skip sends itself into an
* infinite loop.
*
* @author Dave Syer
*
*/
public class SkipPolicyFailedException extends UnexpectedJobExecutionException {
/**
* @param message describes the error to the user
* @param ex the exception that was thrown by a {@link SkipPolicy}
* @param t the exception that caused the skip
*/
public SkipPolicyFailedException(String message, RuntimeException ex, Throwable t) {
super(message + "\n" + t.getClass().getName() + ": " + t.getMessage(), ex);
}
}

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

@@ -34,6 +34,7 @@ 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.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
@@ -202,6 +203,60 @@ public class FaultTolerantStepFactoryBeanTests {
testReadSkip();
}
/**
* Check items causing errors are skipped as expected.
*/
@Test
public void testReadSkipWithPolicyExceptionInReader() throws Exception {
// Should be ignored
factory.setSkipLimit(0);
factory.setSkipPolicy(new SkipPolicy() {
public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
throw new RuntimeException("Planned exception in SkipPolicy");
}
});
reader.setFailures("2");
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals(0, stepExecution.getReadSkipCount());
assertEquals(1, stepExecution.getReadCount());
}
/**
* Check items causing errors are skipped as expected.
*/
@Test
public void testReadSkipWithPolicyExceptionInWriter() throws Exception {
// Should be ignored
factory.setSkipLimit(0);
factory.setSkipPolicy(new SkipPolicy() {
public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException {
throw new RuntimeException("Planned exception in SkipPolicy");
}
});
writer.setFailures("2");
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals(0, stepExecution.getWriteSkipCount());
assertEquals(2, stepExecution.getReadCount());
}
/**
* Check to make sure that ItemStreamException can be skipped. (see
* BATCH-915)

35
spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipPolicyFailedExceptionTests.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* 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.core.step.skip;
import org.springframework.batch.core.listener.AbstractDoubleExceptionTests;
/**
* @author Dave Syer
*
*/
public class SkipPolicyFailedExceptionTests extends AbstractDoubleExceptionTests {
/* (non-Javadoc)
* @see org.springframework.batch.core.listener.AbstractDoubleExceptionTests#getException(java.lang.String, java.lang.RuntimeException, java.lang.Throwable)
*/
@Override
public Exception getException(String msg, RuntimeException cause, Throwable e) throws Exception {
return new SkipPolicyFailedException(msg, cause, e);
}
}