diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java index 8fcfd4cb2..f31bd8d0a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java @@ -23,7 +23,7 @@ package org.springframework.batch.core; * @author Robert Kasanicky * */ -public interface SkipListener extends StepListener { +public interface SkipListener extends StepListener { /** * Callback for a failure on read that is legal, so is not going to be @@ -36,7 +36,7 @@ public interface SkipListener extends StepListener { /** * This item failed on write with the given exception, and a skip was called - * for. The callback occurs immediately after the item is marked for future + * for. The callback occurs immediately after the item is marked for * skipping and is called only once for the same item, regardless of * rollbacks (chunk may be re-processed several times or the exception on * write may not cause rollback at all). @@ -46,4 +46,16 @@ public interface SkipListener extends StepListener { */ void onSkipInWrite(S item, Throwable t); + /** + * This item failed on processing with the given exception, and a skip was called + * for. The callback occurs immediately after the item is marked for + * skipping and is called only once for the same item, regardless of + * rollbacks (chunk may be re-processed several times or the exception on + * write may not cause rollback at all). + * + * @param item the failed item + * @param t the cause of the failure + */ + void onSkipInProcess(T item, Throwable t); + } 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 595cda954..8859c64fe 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 @@ -16,8 +16,8 @@ package org.springframework.batch.core; /** - * Represents a contribution to a {@link StepExecution}, buffering changes until - * they can be applied at a chunk boundary. + * Represents a contribution to a {@link StepExecution}, buffering changes + * until they can be applied at a chunk boundary. * * @author Dave Syer * @@ -26,12 +26,16 @@ public class StepContribution { private volatile int itemCount = 0; + private volatile int filterCount = 0; + private final int parentSkipCount; private volatile int readSkipCount; private volatile int writeSkipCount; + private volatile int processSkipCount; + /** * @param execution */ @@ -39,6 +43,13 @@ public class StepContribution { this.parentSkipCount = execution.getSkipCount(); } + /** + * Increment the counter for the number of items processed. + */ + public void incrementFilterCount(int count) { + filterCount+=count; + } + /** * Increment the counter for the number of items processed. */ @@ -55,12 +66,20 @@ public class StepContribution { return itemCount; } + /** + * Public getter for the filter counter. + * @return the filter counter + */ + public int getFilterCount() { + return filterCount; + } + /** * @return the sum of skips accumulated in the parent {@link StepExecution} * and this StepContribution. */ public int getStepSkipCount() { - return readSkipCount + writeSkipCount + parentSkipCount; + return readSkipCount + writeSkipCount + processSkipCount + parentSkipCount; } /** @@ -69,7 +88,7 @@ public class StepContribution { * parent {@link StepExecution}). */ public int getSkipCount() { - return readSkipCount + writeSkipCount; + return readSkipCount + writeSkipCount + processSkipCount; } /** @@ -93,6 +112,21 @@ public class StepContribution { writeSkipCount++; } + /** + * + */ + public void incrementProcessSkipCount() { + processSkipCount++; + } + + /** + * @param contribution + */ + public void increment(StepContribution contribution) { + itemCount += contribution.getItemCount(); + readSkipCount += contribution.getReadSkipCount(); + } + /** * @return the read skip count */ @@ -107,22 +141,22 @@ public class StepContribution { return writeSkipCount; } + /** + * Public getter for the process skip count. + * @return the process skip count + */ + public int getProcessSkipCount() { + return processSkipCount; + } + /* * (non-Javadoc) * * @see java.lang.Object#toString() */ public String toString() { - return "[StepContribution: items=" + itemCount + ", readSkips=" + readSkipCount - + ", writeSkips=" + writeSkipCount + "]"; - } - - /** - * @param contribution - */ - public void increment(StepContribution contribution) { - itemCount += contribution.getItemCount(); - readSkipCount += contribution.getReadSkipCount(); + return "[StepContribution: items=" + itemCount + "filtered=" + filterCount + ", readSkips=" + readSkipCount + + ", writeSkips=" + writeSkipCount + ", processSkips=" + processSkipCount + "]"; } } 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 382c6bab6..460429b01 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 @@ -63,6 +63,8 @@ public class StepExecution extends Entity { private volatile boolean terminateOnly; + private int filterCount; + /** * Constructor with mandatory properties. * @@ -146,9 +148,9 @@ public class StepExecution extends Entity { } /** - * Returns the current number of items processed for this execution + * Returns the current number of items input for this execution * - * @return the current number of items processed for this execution + * @return the current number of items input for this execution */ public int getItemCount() { return itemCount; @@ -172,6 +174,23 @@ public class StepExecution extends Entity { return rollbackCount; } + /** + * Returns the current number of items filtered out of this execution + * + * @return the current number of items filtered out of this execution + */ + public int getFilterCount() { + return filterCount; + } + + /** + * Public setter for the number of items filtered out of this execution. + * @param filterCount the number of items filtered out of this execution to set + */ + public void setFilterCount(int filterCount) { + this.filterCount = filterCount; + } + /** * Setter for number of rollbacks for this execution */ @@ -278,6 +297,7 @@ public class StepExecution extends Entity { itemCount += contribution.getItemCount(); readSkipCount += contribution.getReadSkipCount(); writeSkipCount += contribution.getWriteSkipCount(); + filterCount += contribution.getFilterCount(); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java index f54f019ae..9db7313be 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java @@ -25,16 +25,16 @@ import org.springframework.core.Ordered; * @author Dave Syer * */ -public class CompositeSkipListener implements SkipListener { +public class CompositeSkipListener implements SkipListener { - private OrderedComposite> listeners = new OrderedComposite>(); + private OrderedComposite> listeners = new OrderedComposite>(); /** * Public setter for the listeners. * * @param listeners */ - public void setListeners(List> listeners) { + public void setListeners(List> listeners) { this.listeners.setItems(listeners); } @@ -43,7 +43,7 @@ public class CompositeSkipListener implements SkipListener { * * @param listener */ - public void register(SkipListener listener) { + public void register(SkipListener listener) { listeners.add(listener); } @@ -53,8 +53,8 @@ public class CompositeSkipListener implements SkipListener { * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable) */ public void onSkipInRead(Throwable t) { - for (Iterator> iterator = listeners.iterator(); iterator.hasNext();) { - SkipListener listener = iterator.next(); + for (Iterator> iterator = listeners.iterator(); iterator.hasNext();) { + SkipListener listener = iterator.next(); listener.onSkipInRead(t); } } @@ -66,9 +66,23 @@ public class CompositeSkipListener implements SkipListener { * java.lang.Throwable) */ public void onSkipInWrite(S item, Throwable t) { - for (Iterator> iterator = listeners.iterator(); iterator.hasNext();) { - SkipListener listener = iterator.next(); + for (Iterator> iterator = listeners.iterator(); iterator.hasNext();) { + SkipListener listener = iterator.next(); listener.onSkipInWrite(item, t); } } + + /** + * Call the registered listeners in order, respecting and prioritising those + * that implement {@link Ordered}. + * @see org.springframework.batch.core.SkipListener#onSkipInWrite(java.lang.Object, + * java.lang.Throwable) + */ + public void onSkipInProcess(T item, Throwable t) { + for (Iterator> iterator = listeners.iterator(); iterator.hasNext();) { + SkipListener listener = iterator.next(); + listener.onSkipInProcess(item, t); + } + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java index f60728374..4aae44061 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java @@ -33,7 +33,7 @@ import org.springframework.batch.repeat.ExitStatus; * */ public class MulticasterBatchListener implements StepExecutionListener, ChunkListener, ItemReadListener, - ItemProcessListener, ItemWriteListener, SkipListener { + ItemProcessListener, ItemWriteListener, SkipListener { private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener(); @@ -45,7 +45,7 @@ public class MulticasterBatchListener implements StepExecutionListener, Ch private CompositeItemWriteListener itemWriteListener = new CompositeItemWriteListener(); - private CompositeSkipListener skipListener = new CompositeSkipListener(); + private CompositeSkipListener skipListener = new CompositeSkipListener(); /** * Initialise the listener instance. @@ -298,4 +298,13 @@ public class MulticasterBatchListener implements StepExecutionListener, Ch skipListener.onSkipInWrite(item, t); } + /** + * @param item + * @param t + * @see org.springframework.batch.core.listener.CompositeSkipListener#onSkipInProcess(Object, Throwable) + */ + public void onSkipInProcess(T item, Throwable t) { + skipListener.onSkipInProcess(item, t); + } + } 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 index c5fabb9b6..3ae689ee6 100644 --- 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 @@ -23,7 +23,7 @@ import org.springframework.batch.core.SkipListener; * @author Dave Syer * */ -public class SkipListenerSupport implements SkipListener { +public class SkipListenerSupport implements SkipListener { /* (non-Javadoc) * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable) @@ -34,8 +34,13 @@ public class SkipListenerSupport implements SkipListener { /* (non-Javadoc) * @see org.springframework.batch.core.SkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable) */ - public void onSkipInWrite(Object item, Throwable t) { + public void onSkipInWrite(S item, Throwable t) { } + /* (non-Javadoc) + * @see org.springframework.batch.core.SkipListener#onSkipInProcess(java.lang.Object, java.lang.Throwable) + */ + public void onSkipInProcess(T item, Throwable t) { + } } 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 6112a50b1..c4c07238d 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 @@ -200,16 +200,19 @@ public class ChunkOrientedTasklet implements Tasklet { * @param contribution current context */ protected void process(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { + int filtered = 0; for (T item : inputs) { - S output = doProcess(item); // TODO: segregate read / write / filter count // (this is read count) contribution.incrementItemCount(); - // TODO: increment filter count if this is null + S output = doProcess(item); if (output != null) { outputs.add(output); + } else { + filtered++; } } + contribution.incrementFilterCount(filtered); inputs.clear(); } 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 4338b3590..e9f0d0f02 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,9 +261,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean exceptions.addAll(new ArrayList>(retryableExceptionClasses)); ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, new ArrayList>(fatalExceptionClasses)); - ChunkOrientedTasklet tasklet = new StatefulRetryTasklet(getItemReader(), - getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, readSkipPolicy, - writeSkipPolicy); + ChunkOrientedTasklet tasklet = new StatefulRetryTasklet(getItemReader(), getItemProcessor(), + getItemWriter(), getChunkOperations(), retryTemplate, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); tasklet.setListeners(getListeners()); step.setTasklet(tasklet); @@ -304,6 +303,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean final private ItemSkipPolicy writeSkipPolicy; + final private ItemSkipPolicy processSkipPolicy; + /** * @param itemReader * @param itemWriter @@ -312,11 +313,12 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean public StatefulRetryTasklet(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations chunkOperations, RetryOperations retryTemplate, ItemSkipPolicy readSkipPolicy, - ItemSkipPolicy writeSkipPolicy) { + ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) { super(itemReader, itemProcessor, itemWriter, chunkOperations); this.retryOperations = retryTemplate; this.readSkipPolicy = readSkipPolicy; this.writeSkipPolicy = writeSkipPolicy; + this.processSkipPolicy = processSkipPolicy; } /** @@ -366,6 +368,79 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean } + /** + * Incorporate retry into the item processor stage. + * + * @see org.springframework.batch.core.step.item.ChunkOrientedTasklet#process(org.springframework.batch.core.StepContribution, + * org.springframework.batch.core.step.item.Chunk, + * org.springframework.batch.core.step.item.Chunk) + */ + @Override + 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(); + + RetryCallback retryCallback = new RetryCallback() { + + public S doWithRetry(RetryContext context) throws Exception { + contribution.incrementItemCount(); + S output = doProcess(item); + return output; + } + + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + public S recover(RetryContext context) throws Exception { + Exception e = context.getLastThrowable(); + if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementProcessSkipCount(); + iterator.remove(e); + } + else { + throw new RetryException("Non-skippable exception in recoverer", 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 + if (output != null) { + outputs.add(output); + } else { + filtered++; + } + + } + + for (Chunk.SkippedItem skip : inputs.getSkips()) { + Exception exception = skip.getException(); + try { + getListener().onSkipInProcess(skip.getItem(), exception); + } + catch (RuntimeException e) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception); + } + } + + contribution.incrementFilterCount(filtered); + + inputs.clear(); + + } + /** * Execute the business logic, delegating to the writer.
* diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/StepContributionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/StepContributionTests.java index 668f2d217..a2161c7b3 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/StepContributionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/StepContributionTests.java @@ -37,4 +37,13 @@ public class StepContributionTests extends TestCase { assertEquals(1, contribution.getItemCount()); } + /** + * Test method for + * {@link org.springframework.batch.core.StepContribution#incrementItemCount()}. + */ + public void testIncrementFilterCount() { + assertEquals(0, contribution.getFilterCount()); + contribution.incrementFilterCount(1); + assertEquals(1, contribution.getFilterCount()); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/StepExecutionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/StepExecutionTests.java index 833c2c31e..c667e3567 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/StepExecutionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/StepExecutionTests.java @@ -119,6 +119,11 @@ public class StepExecutionTests extends TestCase { assertEquals(123, execution.getItemCount()); } + public void testGetFilterCount() { + execution.setFilterCount(123); + assertEquals(123, execution.getFilterCount()); + } + public void testGetJobExecution() throws Exception { assertNotNull(execution.getJobExecution()); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/MulticasterBatchListenerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/MulticasterBatchListenerTests.java index 6c8986237..9c1390ffa 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/MulticasterBatchListenerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/MulticasterBatchListenerTests.java @@ -412,7 +412,7 @@ public class MulticasterBatchListenerTests { */ @Test public void testOnSkipInRead() { - multicast.register(new SkipListenerSupport() { + multicast.register(new SkipListenerSupport() { @Override public void onSkipInRead(Throwable t) { count++; @@ -430,7 +430,7 @@ public class MulticasterBatchListenerTests { */ @Test public void testOnSkipInReadFails() { - multicast.register(new SkipListenerSupport() { + multicast.register(new SkipListenerSupport() { @Override public void onSkipInRead(Throwable t) { count++; @@ -456,7 +456,7 @@ public class MulticasterBatchListenerTests { */ @Test public void testOnSkipInWrite() { - multicast.register(new SkipListenerSupport() { + multicast.register(new SkipListenerSupport() { @Override public void onSkipInWrite(Object item, Throwable t) { count++; @@ -474,7 +474,7 @@ public class MulticasterBatchListenerTests { */ @Test public void testOnSkipInWriteFails() { - multicast.register(new SkipListenerSupport() { + multicast.register(new SkipListenerSupport() { @Override public void onSkipInWrite(Object item, Throwable t) { count++; @@ -493,6 +493,50 @@ public class MulticasterBatchListenerTests { assertEquals(1, count); } + /** + * Test method for + * {@link org.springframework.batch.core.listener.MulticasterBatchListener#onSkipInWrite(java.lang.Object, java.lang.Throwable)} + * . + */ + @Test + public void testOnSkipInProcess() { + multicast.register(new SkipListenerSupport() { + @Override + public void onSkipInProcess(Object item, Throwable t) { + count++; + super.onSkipInWrite(item, t); + } + }); + multicast.onSkipInProcess(null, new RuntimeException("foo")); + assertEquals(1, count); + } + + /** + * Test method for + * {@link org.springframework.batch.core.listener.MulticasterBatchListener#onSkipInWrite(java.lang.Object, java.lang.Throwable)} + * . + */ + @Test + public void testOnSkipInProcessFails() { + multicast.register(new SkipListenerSupport() { + @Override + public void onSkipInProcess(Object item, Throwable t) { + count++; + throw new RuntimeException("foo"); + } + }); + try { + multicast.onSkipInProcess(null, new RuntimeException("bar")); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + // expected + String message = e.getMessage(); + assertEquals("Wrong message: " + message, "foo", message); + } + assertEquals(1, count); + } + /** * @author Dave Syer * diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java index 879f656ef..a5736e083 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java @@ -16,6 +16,7 @@ package org.springframework.batch.core.step.item; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; import java.util.List; @@ -33,6 +34,7 @@ import org.springframework.batch.item.NoWorkFoundException; import org.springframework.batch.item.ParseException; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.support.PassthroughItemProcessor; +import org.springframework.batch.item.validator.ValidationException; import org.springframework.batch.repeat.context.RepeatContextSupport; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; @@ -47,11 +49,11 @@ public class ChunkOrientedTaskletTests { private StubItemReader itemReader = new StubItemReader(); private StubItemWriter itemWriter = new StubItemWriter(); - + private RepeatTemplate repeatTemplate = new RepeatTemplate(); - + private AttributeAccessor context = new RepeatContextSupport(null); - + @Before public void setUp() { repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); @@ -68,6 +70,25 @@ public class ChunkOrientedTaskletTests { assertEquals("12", itemWriter.values); } + @Test + public void testHandleWithItemProcessorFailure() throws Exception { + ChunkOrientedTasklet handler = new ChunkOrientedTasklet(itemReader, + new StubItemProcessor(), itemWriter, repeatTemplate); + StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( + 123L, new JobParameters(), "job")))); + try { + handler.execute(contribution, context); + fail("Expected ValidationException"); + } + catch (ValidationException e) { + // expected + } + assertEquals(2, itemReader.count); + assertEquals(2, contribution.getItemCount()); + assertEquals(0, contribution.getFilterCount()); + assertEquals("", itemWriter.values); + } + @Test public void testHandleCompositeItem() throws Exception { ChunkOrientedTasklet handler = new ChunkOrientedTasklet(itemReader, @@ -76,6 +97,8 @@ public class ChunkOrientedTaskletTests { 123L, new JobParameters(), "job")))); handler.execute(contribution, context); assertEquals(2, itemReader.count); + assertEquals(2, contribution.getItemCount()); + assertEquals(1, contribution.getFilterCount()); assertEquals("12", itemWriter.values); } @@ -100,6 +123,19 @@ public class ChunkOrientedTaskletTests { } } + /** + * @author Dave Syer + * + */ + private static class StubItemProcessor implements ItemProcessor { + public String process(String item) throws Exception { + if ("2".equals(item)) { + throw new ValidationException("Planned failure"); + } + return item; + } + } + /** * @author Dave Syer * @@ -109,7 +145,7 @@ public class ChunkOrientedTaskletTests { public void write(List items) throws Exception { for (String item : items) { - values += item; + values += item; } } } 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 5a2651e2a..628454f54 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 @@ -246,7 +246,7 @@ public class SkipLimitStepFactoryBeanTests { factory.setSkipLimit(3); factory.setItemReader(reader); - factory.setListeners(new StepListener[] { new SkipListenerSupport() { + factory.setListeners(new StepListener[] { new SkipListenerSupport() { @Override public void onSkipInRead(Throwable t) { throw new RuntimeException("oops"); @@ -287,9 +287,9 @@ public class SkipLimitStepFactoryBeanTests { factory.setSkipLimit(3); factory.setItemReader(reader); - factory.setListeners(new StepListener[] { new SkipListenerSupport() { + factory.setListeners(new StepListener[] { new SkipListenerSupport() { @Override - public void onSkipInWrite(Object item, Throwable t) { + public void onSkipInWrite(String item, Throwable t) { throw new RuntimeException("oops"); } } }); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java index 9050334f0..8a980f27b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java @@ -211,8 +211,8 @@ public class StatefulRetryStepFactoryBeanTests { add(RetryException.class); } }); - factory.setListeners(new StepListener[] { new SkipListenerSupport() { - public void onSkipInWrite(Object item, Throwable t) { + factory.setListeners(new StepListener[] { new SkipListenerSupport() { + public void onSkipInWrite(String item, Throwable t) { recovered.add(item); assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); } @@ -275,8 +275,8 @@ public class StatefulRetryStepFactoryBeanTests { add(RetryException.class); } }); - factory.setListeners(new StepListener[] { new SkipListenerSupport() { - public void onSkipInWrite(Object item, Throwable t) { + factory.setListeners(new StepListener[] { new SkipListenerSupport() { + public void onSkipInWrite(String item, Throwable t) { recovered.add(item); assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); } 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 12465ddd8..ac58da53f 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 @@ -55,10 +55,12 @@ public class StatefulRetryTaskletTests { private int limit = 3; - protected int skipLimit = 2; + private int skipLimit = 2; - protected List written = new ArrayList(); + private List written = new ArrayList(); + private List processed = new ArrayList(); + private StatefulRetryTasklet handler; private RepeatTemplate chunkOperations = new RepeatTemplate(); @@ -94,6 +96,7 @@ public class StatefulRetryTaskletTests { private ItemSkipPolicy writeSkipPolicy = readSkipPolicy; + @Before public void setUp() { retryTemplate.setRetryPolicy(new NeverRetryPolicy()); @@ -102,7 +105,7 @@ public class StatefulRetryTaskletTests { @Test public void testBasicHandle() throws Exception { handler = new StatefulRetryTasklet(itemReader, itemProcessor, itemWriter, chunkOperations, - retryTemplate, readSkipPolicy, writeSkipPolicy); + retryTemplate, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); handler.execute(contribution, new BasicAttributeAccessor()); assertEquals(limit, contribution.getItemCount()); @@ -114,7 +117,7 @@ public class StatefulRetryTaskletTests { public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { throw new RuntimeException("Barf!"); } - }, itemProcessor, itemWriter, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy); + }, itemProcessor, itemWriter, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); BasicAttributeAccessor attributes = new BasicAttributeAccessor(); @@ -136,7 +139,7 @@ public class StatefulRetryTaskletTests { written.addAll(items); throw new RuntimeException("Barf!"); } - }, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy); + }, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); BasicAttributeAccessor attributes = new BasicAttributeAccessor(); @@ -155,14 +158,14 @@ public class StatefulRetryTaskletTests { } @Test - public void testSkipMultipleItems() throws Exception { + public void testSkipMultipleItemsOnWrite() throws Exception { handler = new StatefulRetryTasklet(itemReader, itemProcessor, new ItemWriter() { public void write(List items) throws Exception { logger.debug("Writing items: "+items); written.addAll(items); throw new RuntimeException("Barf!"); } - }, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy); + }, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); BasicAttributeAccessor attributes = new BasicAttributeAccessor(); @@ -187,7 +190,7 @@ public class StatefulRetryTaskletTests { attributes = new BasicAttributeAccessor(); try { handler.execute(contribution, attributes); - fail("Expected RuntimeException on i="); + fail("Expected RuntimeException"); } catch (Exception e) { assertEquals("Barf!", e.getMessage()); @@ -205,4 +208,57 @@ public class StatefulRetryTaskletTests { assertEquals(5, written.size()); } + @Test + public void testSkipMultipleItemsOnProcess() 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, 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++) { + try { + handler.execute(contribution, attributes); + fail("Expected RuntimeException on i="+i); + } + catch (Exception e) { + assertEquals("Barf!", e.getMessage()); + } + assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); + } + @SuppressWarnings("unchecked") + Chunk chunk = (Chunk) attributes.getAttribute("INPUT_BUFFER_KEY"); + assertEquals(1, chunk.getSkips().size()); + + // The last recovery for this chunk... + handler.execute(contribution, attributes); + assertEquals(2, chunk.getSkips().size()); + + attributes = new BasicAttributeAccessor(); + try { + handler.execute(contribution, attributes); + fail("Expected RuntimeException"); + } + catch (Exception e) { + assertEquals("Barf!", e.getMessage()); + } + try { + handler.execute(contribution, attributes); + fail("Expected SkipLimitExceededException"); + } + catch (SkipLimitExceededException e) { + // expected + } + assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); + assertEquals(3, contribution.getItemCount()); + assertEquals(2, contribution.getProcessSkipCount()); + assertEquals(3, processed.size()); + } }