From dd46e7c2f69d6040ae77b0cce4e9bcaed4047d5d Mon Sep 17 00:00:00 2001 From: dsyer Date: Sat, 30 Aug 2008 12:16:15 +0000 Subject: [PATCH] BATCH-798: Fixed bug in skip on process (the wrong state was being used for retry). --- .../batch/core/StepContribution.java | 7 ++ .../batch/core/step/item/Chunk.java | 48 +---------- .../core/step/item/ChunkOrientedTasklet.java | 70 ++-------------- .../batch/core/step/item/ItemWrapper.java | 71 ++++++++++++++++ .../step/item/SkipLimitStepFactoryBean.java | 48 +++++------ .../core/step/item/ItemWrapperTests.java | 83 +++++++++++++++++++ .../step/item/StatefulRetryTaskletTests.java | 77 +++++++++++++---- 7 files changed, 255 insertions(+), 149 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java 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 268c38efc..fdc38f9c9 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 @@ -57,6 +57,13 @@ public class StepContribution { itemCount++; } + /** + * Increment the counter for the number of items processed. + */ + public void incrementItemCount(int count) { + itemCount+=count; + } + /** * Public access to the item counter. * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java index aa322cbc0..b0e510fa3 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java @@ -19,7 +19,7 @@ class Chunk implements Iterable { private List items = new ArrayList(); - private List> skips = new ArrayList>(); + private List> skips = new ArrayList>(); /** * Add the item to the chunk. @@ -46,8 +46,8 @@ class Chunk implements Iterable { /** * @return a copy of the skips as an unmodifiable list */ - public List> getSkips() { - return Collections.unmodifiableList(new ArrayList>(skips)); + public List> getSkips() { + return Collections.unmodifiableList(new ArrayList>(skips)); } /** @@ -117,7 +117,7 @@ class Chunk implements Iterable { return; } } - skips.add(new SkippedItem(next, e)); + skips.add(new ItemWrapper(next, e)); iterator.remove(); } @@ -127,44 +127,4 @@ class Chunk implements Iterable { } - /** - * Wrapper for a skipped item and its exception. - * - * @author Dave Syer - * - */ - public static class SkippedItem { - - final private Exception exception; - - final private T item; - - public SkippedItem(T item, Exception e) { - this.item = item; - this.exception = e; - } - - /** - * Public getter for the exception. - * @return the exception - */ - public Exception getException() { - return exception; - } - - /** - * Public getter for the item. - * @return the item - */ - public T getItem() { - return item; - } - - @Override - public String toString() { - return String.format("[exception=%s, item=%s]", exception, item); - } - - } - } \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java index c4c07238d..7a26e1ce1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java @@ -120,7 +120,7 @@ public class ChunkOrientedTasklet implements Tasklet { public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { // TODO: check flags to see if these need to be saved or not (e.g. JMS not) - final Chunk inputs = getInputBuffer(attributes); + final Chunk> inputs = getInputBuffer(attributes); final Chunk outputs = getOutputBuffer(attributes); ExitStatus result = ExitStatus.CONTINUABLE; @@ -134,7 +134,7 @@ public class ChunkOrientedTasklet implements Tasklet { if (item.getItem() == null) { return ExitStatus.FINISHED; } - inputs.add(item.getItem()); + inputs.add(item); return ExitStatus.CONTINUABLE; } }); @@ -199,9 +199,9 @@ public class ChunkOrientedTasklet implements Tasklet { * @param outputs the items to write * @param contribution current context */ - protected void process(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { + protected void process(StepContribution contribution, Chunk> inputs, Chunk outputs) throws Exception { int filtered = 0; - for (T item : inputs) { + for (ItemWrapper item : inputs) { // TODO: segregate read / write / filter count // (this is read count) contribution.incrementItemCount(); @@ -217,11 +217,12 @@ public class ChunkOrientedTasklet implements Tasklet { } /** - * @param item the input item + * @param wrapper the input item * @return the result of the processing * @throws Exception */ - protected S doProcess(T item) throws Exception { + protected S doProcess(ItemWrapper wrapper) throws Exception { + T item = wrapper.getItem(); try { listener.beforeProcess(item); S result = itemProcessor.process(item); @@ -272,7 +273,7 @@ public class ChunkOrientedTasklet implements Tasklet { * @param attributes * @param inputs */ - private void storeInputs(AttributeAccessor attributes, Chunk inputs) { + private void storeInputs(AttributeAccessor attributes, Chunk> inputs) { store(attributes, INPUT_BUFFER_KEY, inputs); } @@ -306,7 +307,7 @@ public class ChunkOrientedTasklet implements Tasklet { } } - private Chunk getInputBuffer(AttributeAccessor attributes) { + private Chunk> getInputBuffer(AttributeAccessor attributes) { return getBuffer(attributes, INPUT_BUFFER_KEY); } @@ -328,57 +329,4 @@ public class ChunkOrientedTasklet implements Tasklet { return resource; } - /** - * @author Dave Syer - * - */ - static protected class ItemWrapper { - - final private T item; - - final private int skipCount; - - /** - * @param item - */ - public ItemWrapper(T item) { - this(item, 0); - } - - /** - * @param item - * @param skipCount - */ - public ItemWrapper(T item, int skipCount) { - this.item = item; - this.skipCount = skipCount; - } - - /** - * @return the item we are wrapping - */ - public T getItem() { - return item; - } - - /** - * Public getter for the skipCount. - * @return the skipCount - */ - public int getSkipCount() { - return skipCount; - } - - /* - * (non-Javadoc) - * - * @see java.lang.Object#toString() - */ - @Override - public String toString() { - return String.format("[%s,%d]", item, skipCount); - } - - } - } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java new file mode 100644 index 000000000..f82b66f0a --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java @@ -0,0 +1,71 @@ +package org.springframework.batch.core.step.item; + +/** + * Wrapper for an item and its exception if it failed processing. + * + * @author Dave Syer + * + */ +public class ItemWrapper { + + final private Exception exception; + + final private T item; + + final private int skipCount; + + /** + * @param item + */ + public ItemWrapper(T item) { + this(item, null, 0); + } + + /** + * @param item + * @param skipCount + */ + public ItemWrapper(T item, int skipCount) { + this(item, null, skipCount); + } + + public ItemWrapper(T item, Exception e) { + this(item, e, 0); + } + + public ItemWrapper(T item, Exception e, int skipCount) { + this.item = item; + this.exception = e; + this.skipCount = skipCount; + } + + /** + * Public getter for the skipCount. + * @return the skipCount + */ + public int getSkipCount() { + return skipCount; + } + + /** + * Public getter for the exception. + * @return the exception + */ + public Exception getException() { + return exception; + } + + /** + * Public getter for the item. + * @return the item + */ + public T getItem() { + return item; + } + + @Override + public String toString() { + return String.format("[exception=%s, item=%s, skips=%d]", exception, item, skipCount); + } + +} \ No newline at end of file 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 be9eb726d..e1dcfc416 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 @@ -261,7 +261,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, new ArrayList>(fatalExceptionClasses)); ChunkOrientedTasklet tasklet = new StatefulRetryTasklet(getItemReader(), getItemProcessor(), - getItemWriter(), getChunkOperations(), retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); + getItemWriter(), getChunkOperations(), retryTemplate, rollbackClassifier, readSkipPolicy, + writeSkipPolicy, writeSkipPolicy); tasklet.setListeners(getListeners()); step.setTasklet(tasklet); @@ -313,7 +314,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean */ public StatefulRetryTasklet(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, - RepeatOperations chunkOperations, RetryOperations retryTemplate, Classifier rollbackClassifier, ItemSkipPolicy readSkipPolicy, + RepeatOperations chunkOperations, RetryOperations retryTemplate, + Classifier rollbackClassifier, ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) { super(itemReader, itemProcessor, itemWriter, chunkOperations); this.retryOperations = retryTemplate; @@ -378,20 +380,19 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean * org.springframework.batch.core.step.item.Chunk) */ @Override - protected void process(final StepContribution contribution, final Chunk inputs, final Chunk outputs) + protected void process(final StepContribution contribution, final Chunk> inputs, final Chunk outputs) throws Exception { int filtered = 0; - for (final Chunk.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { - - final T item = iterator.next(); - + for (final Chunk>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { + + final ItemWrapper wrapper = iterator.next(); + RetryCallback retryCallback = new RetryCallback() { public S doWithRetry(RetryContext context) throws Exception { - contribution.incrementItemCount(); - S output = doProcess(item); + S output = doProcess(wrapper); return output; } @@ -400,43 +401,40 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean RecoveryCallback recoveryCallback = new RecoveryCallback() { public S recover(RetryContext context) throws Exception { - Exception e = context.getLastThrowable(); + Exception e = (Exception) context.getLastThrowable(); if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { contribution.incrementProcessSkipCount(); iterator.remove(e); + return null; } else { - throw new RetryException("Non-skippable exception in recoverer", e); + throw new RetryException("Non-skippable exception in recoverer while processing", e); } - // Unless we reached the end of the chunk we need to rethrow - if (iterator.hasNext()) { - throw e; - } - return null; } - + }; - S output = retryOperations.execute(retryCallback, recoveryCallback, new RetryState(inputs)); - // TODO: increment filter count if this is null + S output = retryOperations.execute(retryCallback, recoveryCallback, new RetryState(wrapper)); if (output != null) { outputs.add(output); - } else { + } + else { filtered++; } } - for (Chunk.SkippedItem skip : inputs.getSkips()) { + for (ItemWrapper> skip : inputs.getSkips()) { Exception exception = skip.getException(); try { - getListener().onSkipInProcess(skip.getItem(), exception); + getListener().onSkipInProcess(skip.getItem().getItem(), exception); } catch (RuntimeException e) { throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception); } } - + + contribution.incrementItemCount(inputs.size()); contribution.incrementFilterCount(filtered); inputs.clear(); @@ -457,8 +455,6 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean RetryCallback retryCallback = new RetryCallback() { public Object doWithRetry(RetryContext context) throws Exception { doWrite(chunk.getItems()); - // TODO: if there is an exception marked as no rollback it - // should get treated as stateless return null; } }; @@ -506,7 +502,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk)); - for (Chunk.SkippedItem skip : chunk.getSkips()) { + for (ItemWrapper skip : chunk.getSkips()) { Exception exception = skip.getException(); try { getListener().onSkipInWrite(skip.getItem(), exception); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java new file mode 100644 index 000000000..2aacaef78 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java @@ -0,0 +1,83 @@ +/* + * 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.item; + +import static org.junit.Assert.*; + +import org.junit.Test; + +/** + * @author Dave Syer + * + */ +public class ItemWrapperTests { + + private Exception exception = new RuntimeException(); + + /** + * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object)}. + */ + @Test + public void testItemWrapperT() { + ItemWrapper wrapper = new ItemWrapper("foo"); + assertEquals("foo", wrapper.getItem()); + assertEquals(null, wrapper.getException()); + assertEquals(0, wrapper.getSkipCount()); + } + + /** + * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, int)}. + */ + @Test + public void testItemWrapperTInt() { + ItemWrapper wrapper = new ItemWrapper("foo",2); + assertEquals("foo", wrapper.getItem()); + assertEquals(null, wrapper.getException()); + assertEquals(2, wrapper.getSkipCount()); + } + + /** + * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, java.lang.Exception)}. + */ + @Test + public void testItemWrapperTException() { + ItemWrapper wrapper = new ItemWrapper("foo",exception); + assertEquals("foo", wrapper.getItem()); + assertEquals(exception, wrapper.getException()); + assertEquals(0, wrapper.getSkipCount()); + } + + /** + * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, java.lang.Exception, int)}. + */ + @Test + public void testItemWrapperTExceptionInt() { + ItemWrapper wrapper = new ItemWrapper("foo", exception, 2); + assertEquals("foo", wrapper.getItem()); + assertEquals(exception , wrapper.getException()); + assertEquals(2, wrapper.getSkipCount()); + } + + /** + * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#toString()}. + */ + @Test + public void testToString() { + ItemWrapper wrapper = new ItemWrapper("foo"); + assertTrue("foo", wrapper.toString().contains("foo")); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryTaskletTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryTaskletTests.java index ed54a7775..f15313166 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryTaskletTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryTaskletTests.java @@ -49,7 +49,7 @@ import org.springframework.batch.support.Classifier; * */ public class StatefulRetryTaskletTests { - + private Log logger = LogFactory.getLog(getClass()); private int count = 0; @@ -61,7 +61,7 @@ public class StatefulRetryTaskletTests { private List written = new ArrayList(); private List processed = new ArrayList(); - + private StatefulRetryTasklet handler; private RepeatTemplate chunkOperations = new RepeatTemplate(); @@ -85,7 +85,7 @@ public class StatefulRetryTaskletTests { }; private RetryTemplate retryTemplate = new RetryTemplate(); - + private Classifier rollbackClassifier = new Classifier() { public Boolean classify(Throwable classifiable) { return true; @@ -103,7 +103,6 @@ public class StatefulRetryTaskletTests { private ItemSkipPolicy writeSkipPolicy = readSkipPolicy; - @Before public void setUp() { retryTemplate.setRetryPolicy(new NeverRetryPolicy()); @@ -124,7 +123,8 @@ public class StatefulRetryTaskletTests { public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { throw new RuntimeException("Barf!"); } - }, itemProcessor, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); + }, itemProcessor, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, + writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); BasicAttributeAccessor attributes = new BasicAttributeAccessor(); @@ -168,7 +168,7 @@ public class StatefulRetryTaskletTests { public void testSkipMultipleItemsOnWrite() throws Exception { handler = new StatefulRetryTasklet(itemReader, itemProcessor, new ItemWriter() { public void write(List items) throws Exception { - logger.debug("Writing items: "+items); + logger.debug("Writing items: " + items); written.addAll(items); throw new RuntimeException("Barf!"); } @@ -176,12 +176,12 @@ public class StatefulRetryTaskletTests { chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); BasicAttributeAccessor attributes = new BasicAttributeAccessor(); - - // Count to 3: (try + skip + skip) + + // Count to 3: (try + skip + skip) for (int i = 0; i < 3; i++) { try { handler.execute(contribution, attributes); - fail("Expected RuntimeException on i="+i); + fail("Expected RuntimeException on i=" + i); } catch (Exception e) { assertEquals("Barf!", e.getMessage()); @@ -216,24 +216,64 @@ public class StatefulRetryTaskletTests { } @Test - public void testSkipMultipleItemsOnProcess() throws Exception { + public void testSkipSingleItemOnProcess() throws Exception { handler = new StatefulRetryTasklet(itemReader, new ItemProcessor() { public String process(Integer item) throws Exception { - logger.debug("Processing item: "+item); + logger.debug("Processing item: " + item); + processed.add(item); + if (item == 3) { + throw new RuntimeException("Barf!"); + } + return "p" + item; + } + }, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, + writeSkipPolicy); + chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3)); + StepContribution contribution = new StepExecution("foo", null).createStepContribution(); + BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + + // try + try { + handler.execute(contribution, attributes); + fail("Expected RuntimeException"); + } + catch (Exception e) { + assertEquals("Barf!", e.getMessage()); + } + assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); + + @SuppressWarnings("unchecked") + Chunk chunk = (Chunk) attributes.getAttribute("INPUT_BUFFER_KEY"); + + // skip... + handler.execute(contribution, attributes); + assertEquals(1, chunk.getSkips().size()); + + assertEquals(2, contribution.getItemCount()); + assertEquals(1, contribution.getProcessSkipCount()); + assertEquals(5, processed.size()); + assertEquals("[p1, p2]", written.toString()); + } + + @Test + public void testSkipOverLimitOnProcess() throws Exception { + handler = new StatefulRetryTasklet(itemReader, new ItemProcessor() { + public String process(Integer item) throws Exception { + logger.debug("Processing item: " + item); processed.add(item); throw new RuntimeException("Barf!"); } - } - , itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); + }, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, + writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); BasicAttributeAccessor attributes = new BasicAttributeAccessor(); - - // Count to 3: (try + skip + try) - for (int i = 0; i < 3; i++) { + + // Count to 3: (try + skip + try) + for (int i = 0; i < 2; i++) { try { handler.execute(contribution, attributes); - fail("Expected RuntimeException on i="+i); + fail("Expected RuntimeException on i=" + i); } catch (Exception e) { assertEquals("Barf!", e.getMessage()); @@ -264,8 +304,9 @@ public class StatefulRetryTaskletTests { // expected } assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); - assertEquals(3, contribution.getItemCount()); + assertEquals(0, contribution.getItemCount()); assertEquals(2, contribution.getProcessSkipCount()); + // Just before the skip at the end we process once more assertEquals(3, processed.size()); } }