From 464ebc577a0429c1ad45b80df41fb9fbcbd06b7a Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 10 Mar 2009 12:10:23 +0000 Subject: [PATCH] OPEN - issue BATCH-1129: Problems with exception classifications http://jira.springframework.org/browse/BATCH-1129 --- .../item/FaultTolerantChunkProcessor.java | 29 +++++++++++- .../skip/NonSkippableProcessException.java | 31 +++++++++++++ .../step/skip/NonSkippableReadException.java | 10 +++-- .../batch/core/step/skip/SkipException.java | 44 +++++++++++++++++++ .../step/skip/SkipLimitExceededException.java | 4 +- .../FaultTolerantChunkProcessorTests.java | 27 ++++++++++-- .../core/step/tasklet/TaskletStepTests.java | 35 ++++----------- .../batch/item/file/FlatFileItemReader.java | 4 +- 8 files changed, 145 insertions(+), 39 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/NonSkippableProcessException.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipException.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index aa25fb965..f0c7bb1ee 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java @@ -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 extends SimpleChunkProcessor retryCallback = new RetryCallback() { 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 extends SimpleChunkProcessor() { + 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.> singleton(DataIntegrityViolationException.class), false)); + Chunk inputs = new Chunk(Arrays.asList("1", "2")); + processor.process(contribution, inputs); + assertEquals(1, list.size()); + } + @Test public void testAfterWrite() throws Exception { Chunk chunk = new Chunk(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); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java index 145a3e7b5..cbef14b8d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java @@ -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; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java index e8efdfa2c..ae3d8225d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java @@ -70,7 +70,7 @@ public class FlatFileItemReader 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 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());