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

OPEN - issue BATCH-1129: Problems with exception classifications

Browse Source 
http://jira.springframework.org/browse/BATCH-1129
This commit is contained in:
dsyer
2009-03-10 12:10:23 +00:00
parent add37da810
commit 464ebc577a
8 changed files with 145 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,6 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.NonSkippableProcessException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
@@ -81,7 +82,31 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
RetryCallback<O> retryCallback = new RetryCallback<O>() {
public O doWithRetry(RetryContext context) throws Exception {
O output = doProcess(item);
O output = null;
try {
output = doProcess(item);
}
catch (Exception e) {
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw e;
}
else if (itemProcessSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// If we are not re-throwing then we should check if
// this is skippable
contribution.incrementProcessSkipCount();
logger.debug("Skipping after failed process with no rollback", e);
}
else {
// If it's not skippable that's an error in
// configuration - it doesn't make sense to not roll
// back if we are also not allowed to skip
throw new NonSkippableProcessException(
"Non-skippable exception in processor. Make sure any exceptions that do not cause a rollback are skippable.",
e);
}
}
if (output == null) {
// No need to re-process filtered items
iterator.remove();
@@ -173,7 +198,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
}
doAfterWrite(outputs.getItems());
contribution.incrementWriteCount(outputs.size());
contribution.incrementWriteCount(outputs.size());
return null;
}

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

@@ -0,0 +1,31 @@
/*
* 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;
/**
* Fatal exception to be thrown when a process operation could not be skipped.
*
* @author Dave Syer
*
*/
public class NonSkippableProcessException extends SkipException {
public NonSkippableProcessException(String msg, Throwable cause) {
super(msg, cause);
}
}

10
spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/NonSkippableReadException.java
View File

@@ -16,9 +16,13 @@
package org.springframework.batch.core.step.skip;
import org.springframework.core.NestedRuntimeException;
public class NonSkippableReadException extends NestedRuntimeException {
/**
* Fatal exception to be thrown when a read operation could not be skipped.
*
* @author Dave Syer
*
*/
public class NonSkippableReadException extends SkipException {
public NonSkippableReadException(String msg, Throwable cause) {
super(msg, cause);

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

@@ -0,0 +1,44 @@
/*
* 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;
/**
* Base exception indicating that the skip has failed or caused a failure.
*
* @author Dave Syer
*/
public abstract class SkipException extends UnexpectedJobExecutionException {
/**
* @param msg the message
* @param nested the cause
*/
public SkipException(String msg, Throwable nested) {
super(msg, nested);
}
/**
* @param msg the message
*/
public SkipException(String msg) {
super(msg);
}
}

4
spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipLimitExceededException.java
View File

@@ -15,7 +15,6 @@
*/
package org.springframework.batch.core.step.skip;
import org.springframework.batch.core.UnexpectedJobExecutionException;
/**
* Exception indicating that the skip limit for a particular {@Step} has
@@ -23,8 +22,9 @@ import org.springframework.batch.core.UnexpectedJobExecutionException;
*
* @author Ben Hale
* @author Lucas Ward
* @author Dave Syer
*/
public class SkipLimitExceededException extends UnexpectedJobExecutionException {
public class SkipLimitExceededException extends SkipException {
private final int skipLimit;

27
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
View File

@@ -5,6 +5,7 @@ import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.junit.Before;
@@ -18,6 +19,8 @@ import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.PassThroughItemProcessor;
import org.springframework.batch.retry.policy.NeverRetryPolicy;
import org.springframework.batch.support.BinaryExceptionClassifier;
import org.springframework.dao.DataIntegrityViolationException;
public class FaultTolerantChunkProcessorTests {
@@ -65,6 +68,21 @@ public class FaultTolerantChunkProcessorTests {
assertEquals(1, list.size());
}
@Test
public void testTransformWithExceptionAndNoRollback() throws Exception {
processor.setItemProcessor(new ItemProcessor<String, String>() {
public String process(String item) throws Exception {
if (item.equals("1")) throw new DataIntegrityViolationException("Planned");
return item;
}
});
processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());
processor.setRollbackClassifier(new BinaryExceptionClassifier(Collections.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class), false));
Chunk<String> inputs = new Chunk<String>(Arrays.asList("1", "2"));
processor.process(contribution, inputs);
assertEquals(1, list.size());
}
@Test
public void testAfterWrite() throws Exception {
Chunk<String> chunk = new Chunk<String>(Arrays.asList("foo", "fail", "bar"));
@@ -78,13 +96,15 @@ public class FaultTolerantChunkProcessorTests {
try {
processor.process(contribution, chunk);
fail();
} catch (RuntimeException e) {
}
catch (RuntimeException e) {
assertEquals("Planned failure!", e.getMessage());
}
try {
processor.process(contribution, chunk);
fail();
} catch (RuntimeException e) {
}
catch (RuntimeException e) {
assertEquals("Planned failure!", e.getMessage());
}
assertEquals(2, chunk.getItems().size());
@@ -120,7 +140,8 @@ public class FaultTolerantChunkProcessorTests {
try {
processor.process(contribution, chunk);
fail();
} catch (RuntimeException e) {
}
catch (RuntimeException e) {
assertEquals("Planned failure!", e.getMessage());
}
processor.process(contribution, chunk);

35
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
View File

@@ -16,7 +16,10 @@
package org.springframework.batch.core.step.tasklet;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.Serializable;
import java.util.ArrayList;
@@ -298,7 +301,7 @@ public class TaskletStepTests {
step.execute(stepExecution);
// context saved before looping and updated once for every processing
// loop (once in this case)
// loop (once in this case)
assertEquals(3, list.size());
}
@@ -543,8 +546,8 @@ public class TaskletStepTests {
step.execute(stepExecution);
assertEquals(BatchStatus.STOPPED, stepExecution.getStatus());
String msg = stepExecution.getExitStatus().getExitDescription();
assertTrue("Message does not contain 'JobInterruptedException': " + msg, contains(msg,
"JobInterruptedException"));
assertTrue("Message does not contain 'JobInterruptedException': " + msg, msg
.contains("JobInterruptedException"));
}
@Test
@@ -619,7 +622,7 @@ public class TaskletStepTests {
step.execute(stepExecution);
assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
String msg = stepExecution.getExitStatus().getExitDescription();
assertTrue("Message does not contain ResetFailedException: " + msg, contains(msg, "ResetFailedException"));
assertTrue("Message does not contain ResetFailedException: " + msg, msg.contains("ResetFailedException"));
// The original rollback was caused by this one:
assertEquals("Bar", stepExecution.getFailureExceptions().get(0).getCause().getMessage());
}
@@ -767,28 +770,6 @@ public class TaskletStepTests {
}
// This doesn't make sense for concurrent tasklet execution scenario.
// @Test
// public void testModifyingExecutionContextMidProcessCausesException() throws Exception {
// StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance));
// final ExecutionContext ec = stepExecution.getExecutionContext();
// step.setTasklet(new Tasklet() {
// public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
// ec.putString("test", "test");
// return RepeatStatus.FINISHED;
// }
// });
//
// step.execute(stepExecution);
// assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// assertEquals(1, stepExecution.getFailureExceptions().size());
// assertTrue(stepExecution.getFailureExceptions().get(0) instanceof IllegalStateException);
// }
private boolean contains(String str, String searchStr) {
return str.indexOf(searchStr) != -1;
}
private static class JobRepositoryStub extends JobRepositorySupport {
private int updateCount = 0;

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java
View File

@@ -70,7 +70,7 @@ public class FlatFileItemReader<T> extends AbstractItemCountingItemStreamItemRea
private LineCallbackHandler skippedLinesCallback;
private boolean strict = false;
private boolean strict = true;
public FlatFileItemReader() {
setName(ClassUtils.getShortName(FlatFileItemReader.class));
@@ -243,7 +243,7 @@ public class FlatFileItemReader<T> extends AbstractItemCountingItemStreamItemRea
noInput = false;
if (!resource.exists()) {
if (strict) {
throw new IllegalStateException("Input resource must exist (reader is in 'strict' mode)");
throw new IllegalStateException("Input resource must exist (reader is in 'strict' mode): "+resource);
}
noInput = true;
logger.warn("Input resource does not exist " + resource.getDescription());