diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java index 4ba429abe..922c3e7c4 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java @@ -15,7 +15,6 @@ */ package org.springframework.batch.core; - /** * Represents a contribution to a {@link StepExecution}, buffering changes * until they can be applied at a chunk boundary. @@ -84,12 +83,31 @@ public class StepContribution { * StepContribution (not including skips accumulated in the * parent {@link StepExecution}. */ - public int getContributionSkipCount() { + public int getSkipCount() { return skipCount; } + /** + * Increment the skip count + */ public void incrementSkipCount() { skipCount++; } + /** + * Increment the skip count by a non-trivial amount + * @param delta + */ + public void incrementSkipCount(int delta) { + skipCount += delta; + } + + /* + * (non-Javadoc) + * @see java.lang.Object#toString() + */ + public String toString() { + return "[StepContribution: items="+itemCount+", commits=" + commitCount + ", skips=" + skipCount + "]"; + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java index d8c434438..e40c6eb30 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java @@ -307,7 +307,7 @@ public class StepExecution extends Entity { // TODO: this should not be necessary - the step decides // executionContext = contribution.getExecutionContext(); commitCount += contribution.getCommitCount(); - skipCount += contribution.getContributionSkipCount(); + skipCount += contribution.getSkipCount(); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/SkipListenerSupport.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/SkipListenerSupport.java new file mode 100644 index 000000000..c5fabb9b6 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/SkipListenerSupport.java @@ -0,0 +1,41 @@ +/* + * Copyright 2006-2008 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.listener; + +import org.springframework.batch.core.SkipListener; + +/** + * Basic no-op implementations of all {@link SkipListener} implementations. + * + * @author Dave Syer + * + */ +public class SkipListenerSupport implements SkipListener { + + /* (non-Javadoc) + * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable) + */ + public void onSkipInRead(Throwable t) { + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.SkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable) + */ + public void onSkipInWrite(Object item, Throwable t) { + } + + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java index a48f231bc..0190f5f91 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java @@ -330,12 +330,12 @@ public class ItemOrientedStep extends AbstractStep { } catch (Error e) { - stepExecution.incrementSkipCountBy(contribution.getContributionSkipCount()); + stepExecution.incrementSkipCountBy(contribution.getSkipCount()); processRollback(stepExecution, fatalException, transaction); throw e; } catch (Exception e) { - stepExecution.incrementSkipCountBy(contribution.getContributionSkipCount()); + stepExecution.incrementSkipCountBy(contribution.getSkipCount()); processRollback(stepExecution, fatalException, transaction); throw e; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index 7afe64e45..2004b1cc7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -1,9 +1,12 @@ package org.springframework.batch.core.step.item; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; +import org.springframework.batch.core.step.skip.SkipLimitExceededException; import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler; @@ -92,14 +95,24 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * to absorb exceptions at the step level because the failed items * will never re-appear after a rollback. */ - itemHandler.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(skipLimit, Arrays - .asList(skippableExceptionClasses), Arrays.asList(fatalExceptionClasses))); - SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler(); - exceptionHandler.setLimit(skipLimit); + List fatalExceptionList = new ArrayList(Arrays.asList(fatalExceptionClasses)); + if (!fatalExceptionList.contains(SkipLimitExceededException.class)) { + fatalExceptionList.add(SkipLimitExceededException.class); + } + fatalExceptionClasses = (Class[]) fatalExceptionList.toArray(new Class[0]); + + LimitCheckingItemSkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, Arrays + .asList(skippableExceptionClasses), fatalExceptionList); + itemHandler.setItemSkipPolicy(skipPolicy); + SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler(skipLimit); exceptionHandler.setExceptionClasses(skippableExceptionClasses); exceptionHandler.setFatalExceptionClasses(fatalExceptionClasses); + + getStepOperations().setExceptionHandler(exceptionHandler); + + // for subclass to pick up limit and exception classes setExceptionHandler(exceptionHandler); - getStepOperations().setExceptionHandler(getExceptionHandler()); + itemHandler.setItemKeyGenerator(itemKeyGenerator); BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java index e10dd86bf..dfc2836fe 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java @@ -68,7 +68,7 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { catch (SkippableException e) { // expected } - assertEquals(0, contribution.getContributionSkipCount()); + assertEquals(0, contribution.getSkipCount()); assertEquals(new Holder("3"), handler.read(contribution)); } @@ -76,7 +76,7 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); assertEquals(new Holder("1"), handler.read(contribution)); assertEquals(new Holder("3"), handler.read(contribution)); - assertEquals(1, contribution.getContributionSkipCount()); + assertEquals(1, contribution.getSkipCount()); assertEquals(new Holder("4"), handler.read(contribution)); } @@ -89,14 +89,14 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { catch (SkippableException e) { // expected } - assertEquals(0, contribution.getContributionSkipCount()); + assertEquals(0, contribution.getSkipCount()); } public void testHandleWithSkip() throws Exception { handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); handler.handle(contribution); handler.handle(contribution); - assertEquals(1, contribution.getContributionSkipCount()); + assertEquals(1, contribution.getSkipCount()); // 2 is skipped so 3 was last one processed and now we are at 4 try { handler.handle(contribution); @@ -105,7 +105,7 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { catch (SkippableException e) { // expected } - assertEquals(2, contribution.getContributionSkipCount()); + assertEquals(2, contribution.getSkipCount()); // No "4" because it was skipped on write assertEquals(new Holder("5"), handler.read(contribution)); } @@ -121,11 +121,11 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { } handler.handle(contribution); handler.handle(contribution); - assertEquals(2, contribution.getContributionSkipCount()); + assertEquals(2, contribution.getSkipCount()); // 2 is skipped so 3 was last one processed and now we are at 4, which was previously skipped handler.handle(contribution); assertEquals(null, handler.read(contribution)); - assertEquals(2, contribution.getContributionSkipCount()); + assertEquals(2, contribution.getSkipCount()); assertEquals(1, TransactionSynchronizationManager.getResourceMap().size()); Set removed = (Set) TransactionSynchronizationManager.getResourceMap().values().iterator().next(); @@ -151,10 +151,10 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { catch (SkippableException e) { // expected } - assertEquals(1, contribution.getContributionSkipCount()); + assertEquals(1, contribution.getSkipCount()); assertEquals(new Holder("1"), handler.read(contribution)); assertEquals(new Holder("3"), handler.read(contribution)); - assertEquals(2, contribution.getContributionSkipCount()); + assertEquals(2, contribution.getSkipCount()); // No "4" because it was skipped on write assertEquals(new Holder("5"), handler.read(contribution)); } @@ -165,7 +165,7 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); handler.handle(contribution); handler.handle(contribution); - assertEquals(1, contribution.getContributionSkipCount()); + assertEquals(1, contribution.getSkipCount()); // 2 is skipped so 3 was last one processed and now we are at 4 try { handler.handle(contribution); @@ -174,7 +174,7 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { catch (SkippableException e) { // expected } - assertEquals(2, contribution.getContributionSkipCount()); + assertEquals(2, contribution.getSkipCount()); // No "4" because it was skipped on write, even though it is mutating // its key assertEquals(new Holder("5"), handler.read(contribution)); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java index 386af4ae9..91f7b16a9 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java @@ -1,7 +1,9 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import junit.framework.TestCase; @@ -12,6 +14,7 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.JobSupport; import org.springframework.batch.core.step.JobRepositorySupport; +import org.springframework.batch.core.step.skip.SkipLimitExceededException; import org.springframework.batch.item.ClearFailedException; import org.springframework.batch.item.FlushFailedException; import org.springframework.batch.item.ItemReader; @@ -22,13 +25,14 @@ import org.springframework.batch.item.ParseException; import org.springframework.batch.item.ResetFailedException; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.util.StringUtils; /** * Tests for {@link SkipLimitStepFactoryBean}. */ public class SkipLimitStepFactoryBeanTests extends TestCase { - SkipLimitStepFactoryBean tested = new SkipLimitStepFactoryBean(); + SkipLimitStepFactoryBean factory = new SkipLimitStepFactoryBean(); Class[] skippableExceptions = new Class[] { SkippableException.class, SkippableRuntimeException.class }; @@ -43,13 +47,13 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { JobExecution jobExecution; protected void setUp() throws Exception { - tested.setJobRepository(new JobRepositorySupport()); - tested.setTransactionManager(new ResourcelessTransactionManager()); - tested.setCommitInterval(COMMIT_INTERVAL); - tested.setItemReader(reader); - tested.setItemWriter(writer); - tested.setSkippableExceptionClasses(skippableExceptions); - tested.setSkipLimit(SKIP_LIMIT); + factory.setJobRepository(new JobRepositorySupport()); + factory.setTransactionManager(new ResourcelessTransactionManager()); + factory.setCommitInterval(COMMIT_INTERVAL); + factory.setItemReader(reader); + factory.setItemWriter(writer); + factory.setSkippableExceptionClasses(skippableExceptions); + factory.setSkipLimit(SKIP_LIMIT); JobInstance jobInstance = new JobInstance(new Long(1), new JobParameters(), new JobSupport("skipJob")); jobExecution = new JobExecution(jobInstance); @@ -59,7 +63,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { * Check items causing errors are skipped as expected. */ public void testSkip() throws Exception { - ItemOrientedStep step = (ItemOrientedStep) tested.getObject(); + ItemOrientedStep step = (ItemOrientedStep) factory.getObject(); StepExecution stepExecution = new StepExecution(step, jobExecution); step.execute(stepExecution); @@ -70,13 +74,8 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { assertTrue(reader.processed.contains("4")); assertFalse(writer.written.contains("4")); - String[] expectedOutput = { "1", "3", "5" }; - - for (int i = 0; i < expectedOutput.length; i++) { - assertTrue("Output should contain \"" + expectedOutput[i] + "\"", writer.written - .contains(expectedOutput[i])); - } - assertTrue(writer.written.size() == expectedOutput.length); + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,3,5")); + assertEquals(expectedOutput, writer.written); } @@ -85,14 +84,14 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { * skip settings (note the fatal exception is also classified as skippable). */ public void testFatalException() throws Exception { - tested.setFatalExceptionClasses(new Class[] { FatalRuntimeException.class }); - tested.setItemWriter(new SkipWriterStub() { + factory.setFatalExceptionClasses(new Class[] { FatalRuntimeException.class }); + factory.setItemWriter(new SkipWriterStub() { public void write(Object item) { throw new FatalRuntimeException("Ouch!"); } }); - ItemOrientedStep step = (ItemOrientedStep) tested.getObject(); + ItemOrientedStep step = (ItemOrientedStep) factory.getObject(); StepExecution stepExecution = new StepExecution(step, jobExecution); try { @@ -104,26 +103,105 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { } } + /** + * Check items causing errors are skipped as expected. + */ + public void testSkipOverLimit() throws Exception { + + factory.setSkipLimit(1); + + ItemOrientedStep step = (ItemOrientedStep) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step, jobExecution); + + try { + step.execute(stepExecution); + fail("Expected SkipLimitExceededException."); + } + catch (SkipLimitExceededException e) { + } + + assertEquals(1, stepExecution.getSkipCount()); + + // writer did not skip "2" as it never made it to writer, only "4" did + assertTrue(reader.processed.contains("4")); + assertFalse(writer.written.contains("4")); + + // failure on "4" tripped the skip limit so we never got to "5" + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,3")); + assertEquals(expectedOutput, writer.written); + + } + + /** + * Check items causing errors are skipped as expected. + */ + public void testSkipOverLimitOnRead() throws Exception { + + reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils + .commaDelimitedListToSet("2,3,5")); + + factory.setSkipLimit(3); + factory.setItemReader(reader); + factory.setSkippableExceptionClasses(new Class[] {Exception.class}); + + ItemOrientedStep step = (ItemOrientedStep) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step, jobExecution); + + try { + step.execute(stepExecution); + fail("Expected SkipLimitExceededException."); + } + catch (SkipLimitExceededException e) { + } + + assertEquals(3, stepExecution.getSkipCount()); + + // writer did not skip "2" as it never made it to writer, only "4" did + assertTrue(reader.processed.contains("4")); + + // failure on "4" tripped the skip limit so we never got to "5" + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1")); + assertEquals(expectedOutput, writer.written); + + } + /** * Simple item reader that supports skip functionality. */ private static class SkipReaderStub implements ItemReader { - final String[] items = { "1", "2", "3", "4", "5", null }; + private final String[] items; - Collection processed = new ArrayList(); + private Collection processed = new ArrayList(); - int counter = -1; + private int counter = -1; - int marked = 0; + private int marked = 0; + + private final Collection failures; + + public SkipReaderStub() { + this(new String[] { "1", "2", "3", "4", "5" }, Collections.singleton("2")); + } + + public SkipReaderStub(String[] items, Collection failures) { + this.items = items; + this.failures = failures; + } public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { counter++; - if ("2".equals(items[counter])) { + if (counter >= items.length) { + return null; + } + String item = items[counter]; + if (failures.contains(item)) { throw new SkippableException("exception in reader"); } - processed.add(items[counter]); - return items[counter]; + processed.add(item); + return item; } public void mark() throws MarkFailedException {