From 223a58da185a71ba0c0b367ee948dbf48b17c2cc Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 28 Aug 2008 07:56:38 +0000 Subject: [PATCH] OPEN - issue BATCH-220: Chunk-oriented approach to processing Add small optimisation to skip chunk with single item --- .../batch/core/ItemProcessListener.java | 29 +++ .../CompositeItemProcessListener.java | 88 +++++++ .../listener/MulticasterBatchListener.java | 53 ++++- .../step/handler/BasicAttributeAccessor.java | 10 + .../core/step/handler/StepHandlerStep.java | 8 - .../batch/core/step/item/Chunk.java | 17 +- .../step/item/ItemOrientedStepHandler.java | 220 ++++++++++-------- .../step/item/SkipLimitStepFactoryBean.java | 29 ++- .../CompositeItemProcessListenerTests.java | 88 +++++++ .../StatefulRetryStepFactoryBeanTests.java | 24 +- .../item/StatefulRetryStepHandlerTests.java | 208 +++++++++++++++++ .../support/RetrySampleItemWriterTests.java | 4 +- 12 files changed, 647 insertions(+), 131 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/ItemProcessListener.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeItemProcessListener.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/BasicAttributeAccessor.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeItemProcessListenerTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepHandlerTests.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/ItemProcessListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/ItemProcessListener.java new file mode 100644 index 000000000..0256b4130 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/ItemProcessListener.java @@ -0,0 +1,29 @@ +/* + * 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; + +/** + * @author Dave Syer + * + */ +public interface ItemProcessListener extends StepListener { + + void beforeProcess(T item); + + void afterProcess(T item, S result); + + void onProcessError(T item, Exception e); +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeItemProcessListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeItemProcessListener.java new file mode 100644 index 000000000..a0b05dec1 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeItemProcessListener.java @@ -0,0 +1,88 @@ +/* + * 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.listener; + +import java.util.Iterator; +import java.util.List; + +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.core.Ordered; + +/** + * @author Dave Syer + * + */ +public class CompositeItemProcessListener implements ItemProcessListener { + + private OrderedComposite> listeners = new OrderedComposite>(); + + /** + * Public setter for the listeners. + * + * @param itemReadListeners + */ + public void setListeners(List> itemReadListeners) { + this.listeners.setItems(itemReadListeners); + } + + /** + * Register additional listener. + * + * @param itemReaderListener + */ + public void register(ItemProcessListener itemReaderListener) { + listeners.add(itemReaderListener); + } + + /** + * Call the registered listeners in reverse order, respecting and + * prioritising those that implement {@link Ordered}. + * @see org.springframework.batch.core.ItemProcessListener#afterProcess(java.lang.Object, + * java.lang.Object) + */ + public void afterProcess(T item, S result) { + for (Iterator> iterator = listeners.reverse(); iterator.hasNext();) { + ItemProcessListener listener = iterator.next(); + listener.afterProcess(item, result); + } + } + + /** + * Call the registered listeners in order, respecting and prioritising those + * that implement {@link Ordered}. + * @see org.springframework.batch.core.ItemProcessListener#beforeProcess(java.lang.Object) + */ + public void beforeProcess(T item) { + for (Iterator> iterator = listeners.iterator(); iterator.hasNext();) { + ItemProcessListener listener = iterator.next(); + listener.beforeProcess(item); + } + } + + /** + * Call the registered listeners in reverse order, respecting and + * prioritising those that implement {@link Ordered}. + * @see org.springframework.batch.core.ItemProcessListener#onProcessError(java.lang.Object, + * java.lang.Exception) + */ + public void onProcessError(T item, Exception e) { + for (Iterator> iterator = listeners.reverse(); iterator.hasNext();) { + ItemProcessListener listener = iterator.next(); + listener.onProcessError(item, e); + } + } + +} 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 d4635dbb1..f60728374 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 @@ -18,6 +18,7 @@ package org.springframework.batch.core.listener; import java.util.List; import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.ItemProcessListener; import org.springframework.batch.core.ItemReadListener; import org.springframework.batch.core.ItemWriteListener; import org.springframework.batch.core.SkipListener; @@ -32,7 +33,7 @@ import org.springframework.batch.repeat.ExitStatus; * */ public class MulticasterBatchListener implements StepExecutionListener, ChunkListener, ItemReadListener, - ItemWriteListener, SkipListener { + ItemProcessListener, ItemWriteListener, SkipListener { private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener(); @@ -40,6 +41,8 @@ public class MulticasterBatchListener implements StepExecutionListener, Ch private CompositeItemReadListener itemReadListener = new CompositeItemReadListener(); + private CompositeItemProcessListener itemProcessListener = new CompositeItemProcessListener(); + private CompositeItemWriteListener itemWriteListener = new CompositeItemWriteListener(); private CompositeSkipListener skipListener = new CompositeSkipListener(); @@ -76,10 +79,13 @@ public class MulticasterBatchListener implements StepExecutionListener, Ch this.chunkListener.register((ChunkListener) listener); } if (listener instanceof ItemReadListener) { + // TODO: make this type safe somehow? this.itemReadListener.register((ItemReadListener) listener); } + if (listener instanceof ItemProcessListener) { + this.itemProcessListener.register((ItemProcessListener) listener); + } if (listener instanceof ItemWriteListener) { - // TODO: make this type safe somehow? this.itemWriteListener.register((ItemWriteListener) listener); } if (listener instanceof SkipListener) { @@ -87,6 +93,49 @@ public class MulticasterBatchListener implements StepExecutionListener, Ch } } + /** + * @param item + * @param result + * @see org.springframework.batch.core.listener.CompositeItemProcessListener#afterProcess(java.lang.Object, + * java.lang.Object) + */ + public void afterProcess(T item, S result) { + try { + itemProcessListener.afterProcess(item, result); + } + catch (RuntimeException e) { + throw new StepListenerFailedException("Error in afterProcess.", e); + } + } + + /** + * @param item + * @see org.springframework.batch.core.listener.CompositeItemProcessListener#beforeProcess(java.lang.Object) + */ + public void beforeProcess(T item) { + try { + itemProcessListener.beforeProcess(item); + } + catch (RuntimeException e) { + throw new StepListenerFailedException("Error in beforeProcess.", e); + } + } + + /** + * @param item + * @param ex + * @see org.springframework.batch.core.listener.CompositeItemProcessListener#onProcessError(java.lang.Object, + * java.lang.Exception) + */ + public void onProcessError(T item, Exception ex) { + try { + itemProcessListener.onProcessError(item, ex); + } + catch (RuntimeException e) { + throw new StepListenerFailedException("Error in onProcessError.", e); + } + } + /** * @see org.springframework.batch.core.listener.CompositeStepExecutionListener#afterStep(StepExecution) */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/BasicAttributeAccessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/BasicAttributeAccessor.java new file mode 100644 index 000000000..bc9e51632 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/BasicAttributeAccessor.java @@ -0,0 +1,10 @@ +package org.springframework.batch.core.step.handler; + +import org.springframework.core.AttributeAccessorSupport; + +/** + * @author Dave Syer + * + */ +public class BasicAttributeAccessor extends AttributeAccessorSupport { +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java index 81ddea5ef..d5a68fff2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java @@ -42,7 +42,6 @@ import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.core.AttributeAccessor; -import org.springframework.core.AttributeAccessorSupport; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; @@ -386,13 +385,6 @@ public class StepHandlerStep extends AbstractStep { } } - /** - * @author Dave Syer - * - */ - private static final class BasicAttributeAccessor extends AttributeAccessorSupport { - } - private static class ExceptionHolder { private Exception exception; 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 85a499263..aa322cbc0 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 @@ -65,6 +65,13 @@ class Chunk implements Iterable { return new ChunkIterator(items); } + /** + * @return the number of items (excluding skips) + */ + public int size() { + return items.size(); + } + /* * (non-Javadoc) * @@ -102,9 +109,15 @@ class Chunk implements Iterable { } public void remove(Exception e) { - if (next != null) { - skips.add(new SkippedItem(next, e)); + if (next == null) { + if (iterator.hasNext()) { + next = iterator.next(); + } + else { + return; + } } + skips.add(new SkippedItem(next, e)); iterator.remove(); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java index d0e1b58fe..0bf0ba10b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java @@ -78,6 +78,36 @@ public class ItemOrientedStepHandler implements StepHandler { this.repeatOperations = repeatOperations; } + /** + * Register some {@link StepListener}s with the handler. Each will get + * the callbacks in the order specified at the correct stage. + * + * @param listeners + */ + public void setListeners(StepListener[] listeners) { + for (StepListener listener : listeners) { + registerListener(listener); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a + * process. + * + * @param listener a {@link StepListener} + */ + public void registerListener(StepListener listener) { + this.listener.register(listener); + } + + /** + * Public getter for the listener. + * @return the listener + */ + protected MulticasterBatchListener getListener() { + return listener; + } + /** * Get the next item from {@link #read(StepContribution)} and if not null * pass the item to {@link #write(Chunk, StepContribution)}. If the @@ -117,21 +147,9 @@ public class ItemOrientedStepHandler implements StepHandler { storeInputs(attributes, inputs); } - - for (T item : inputs) { - - // TODO: processor listener - S output = itemProcessor.process(item); - - // TODO: segregate read / write / filter count - // (this is read count) - contribution.incrementItemCount(); - - // TODO: increment filter count if this is null - if (output != null) { - outputs.add(output); - } - + + if (!inputs.isEmpty()) { + process(contribution, inputs, outputs); } storeOutputsAndClearInputs(attributes, outputs, contribution); @@ -150,6 +168,96 @@ public class ItemOrientedStepHandler implements StepHandler { } + /** + * @param contribution current context + * @return next item for writing + */ + protected ItemWrapper read(StepContribution contribution) throws Exception { + return new ItemWrapper(doRead()); + } + + /** + * @return item + * @throws Exception + */ + protected final T doRead() throws Exception { + try { + listener.beforeRead(); + T item = itemReader.read(); + listener.afterRead(item); + return item; + } + catch (Exception e) { + listener.onReadError(e); + throw e; + } + } + + /** + * + * @param inputs the items to process + * @param outputs the items to write + * @param contribution current context + */ + protected void process(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { + 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 + if (output != null) { + outputs.add(output); + } + } + inputs.clear(); + } + + /** + * @param item the input item + * @return the result of the processing + * @throws Exception + */ + protected S doProcess(T item) throws Exception { + try { + listener.beforeProcess(item); + S result = itemProcessor.process(item); + listener.afterProcess(item, result); + return result; + } + catch (Exception e) { + listener.onProcessError(item, e); + throw e; + } + } + + /** + * + * @param chunk the items to write + * @param contribution current context + */ + protected void write(Chunk chunk, StepContribution contribution) throws Exception { + doWrite(chunk.getItems()); + chunk.clear(); + } + + /** + * @param items + * @throws Exception + */ + protected final void doWrite(List items) throws Exception { + try { + listener.beforeWrite(items); + itemWriter.write(items); + // TODO: increment write count + listener.afterWrite(items); + } + catch (Exception e) { + listener.onWriteError(e, items); + throw e; + } + } + /** * @param attributes */ @@ -217,88 +325,6 @@ public class ItemOrientedStepHandler implements StepHandler { return resource; } - /** - * @param contribution current context - * @return next item for writing - */ - protected ItemWrapper read(StepContribution contribution) throws Exception { - return new ItemWrapper(doRead()); - } - - /** - * @return item - * @throws Exception - */ - protected final T doRead() throws Exception { - try { - listener.beforeRead(); - T item = itemReader.read(); - listener.afterRead(item); - return item; - } - catch (Exception e) { - listener.onReadError(e); - throw e; - } - } - - /** - * - * @param chunk the items to write - * @param contribution current context - */ - protected void write(Chunk chunk, StepContribution contribution) throws Exception { - doWrite(chunk.getItems()); - chunk.clear(); - } - - /** - * @param items - * @throws Exception - */ - protected final void doWrite(List items) throws Exception { - try { - listener.beforeWrite(items); - itemWriter.write(items); - // TODO: increment write count - listener.afterWrite(items); - } - catch (Exception e) { - listener.onWriteError(e, items); - throw e; - } - } - - /** - * Register some {@link StepListener}s with the handler. Each will get - * the callbacks in the order specified at the correct stage. - * - * @param listeners - */ - public void setListeners(StepListener[] listeners) { - for (StepListener listener : listeners) { - registerListener(listener); - } - } - - /** - * Register a listener for callbacks at the appropriate stages in a - * process. - * - * @param listener a {@link StepListener} - */ - public void registerListener(StepListener listener) { - this.listener.register(listener); - } - - /** - * Public getter for the listener. - * @return the listener - */ - protected MulticasterBatchListener getListener() { - return listener; - } - /** * @author Dave Syer * 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 d811eb36b..c7afc42ec 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 @@ -390,7 +390,13 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean public Object recover(RetryContext context) throws Exception { - Exception t = (Exception) context.getLastThrowable(); + // small optimisation: if there was only one item, then we + // don't have to try writing it again to see if it fails... + if (chunk.size() == 1) { + Exception e = (Exception) context.getLastThrowable(); + checkSkipPolicy(contribution, chunk.iterator(), e); + return null; + } for (Chunk.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) { S item = iterator.next(); @@ -398,20 +404,25 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean doWrite(Collections.singletonList(item)); } catch (Exception e) { - if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { - iterator.remove(e); - contribution.incrementWriteSkipCount(); - throw e; - } - else { - throw new RetryException("Non-skippable exception in recoverer", t); - } + checkSkipPolicy(contribution, iterator, e); + throw e; } } return null; } + + private void checkSkipPolicy(final StepContribution contribution, Chunk.ChunkIterator iterator, + Exception e) throws Exception { + if (writeSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementWriteSkipCount(); + iterator.remove(e); + } + else { + throw new RetryException("Non-skippable exception in recoverer", e); + } + } }; retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk)); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeItemProcessListenerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeItemProcessListenerTests.java new file mode 100644 index 000000000..ba695c29e --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeItemProcessListenerTests.java @@ -0,0 +1,88 @@ +/* + * 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 static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; + +import java.util.ArrayList; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.ItemProcessListener; + +/** + * @author Dave Syer + * + */ +public class CompositeItemProcessListenerTests { + + private ItemProcessListener listener; + + private CompositeItemProcessListener compositeListener; + + @SuppressWarnings("unchecked") + @Before + public void setUp() throws Exception { + listener = createMock(ItemProcessListener.class); + compositeListener = new CompositeItemProcessListener(); + compositeListener.register(listener); + } + + @Test + public void testBeforeRProcess() { + Object item = new Object(); + listener.beforeProcess(item); + replay(listener); + compositeListener.beforeProcess(item); + verify(listener); + } + + @Test + public void testAfterRead() { + Object item = new Object(); + Object result = new Object(); + listener.afterProcess(item, result); + replay(listener); + compositeListener.afterProcess(item, result); + verify(listener); + } + + @Test + public void testOnReadError() { + Object item = new Object(); + Exception ex = new Exception(); + listener.onProcessError(item, ex); + replay(listener); + compositeListener.onProcessError(item, ex); + verify(listener); + } + + @Test + public void testSetListeners() throws Exception { + compositeListener.setListeners(new ArrayList>() { + { + add(listener); + } + }); + listener.beforeProcess(null); + replay(listener); + compositeListener.beforeProcess(null); + verify(listener); + } + +} 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 e2bcfd051..9050334f0 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 @@ -73,7 +73,7 @@ public class StatefulRetryStepFactoryBeanTests { private List provided = new ArrayList(); private List written = TransactionAwareProxyFactory.createTransactionalList(); - + int count = 0; private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), @@ -260,8 +260,8 @@ public class StatefulRetryStepFactoryBeanTests { // [a, b, c, d, e, f, null] assertEquals(7, provided.size()); - // [a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f] - assertEquals(16, processed.size()); + // [a, b, b, b, b, b, c, d, d, d, d, d, e, f] + assertEquals(14, processed.size()); // [b, d] assertEquals(2, recovered.size()); } @@ -324,7 +324,8 @@ public class StatefulRetryStepFactoryBeanTests { // [a, b, c, d, e, f, null] assertEquals(7, provided.size()); - // [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, a, c, d, e, f, d, e, f, d, e, f, d, e, f, d, e, f, d, e, f] + // [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, a, c, d, e, f, d, + // e, f, d, e, f, d, e, f, d, e, f, d, e, f] assertEquals(37, processed.size()); // [b, d] assertEquals(2, recovered.size()); @@ -374,9 +375,10 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(0, stepExecution.getSkipCount()); // [b] assertEquals(1, provided.size()); - // the failed items are tried one more time than the limit (TODO: maybe fix this?) - // [b, b, b, b, b] - assertEquals(5, processed.size()); + // the failed items are tried up to the limit (but only precisely so if + // the commit interval is 1) + // [b, b, b, b] + assertEquals(4, processed.size()); // [] assertEquals(0, recovered.size()); assertEquals(1, stepExecution.getItemCount()); @@ -432,8 +434,8 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(0, stepExecution.getSkipCount()); // [b] assertEquals(1, provided.size()); - // [b, b] - assertEquals(2, processed.size()); + // [b] + assertEquals(1, processed.size()); // [] assertEquals(0, recovered.size()); assertEquals(1, stepExecution.getItemCount()); @@ -478,8 +480,8 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(0, stepExecution.getSkipCount()); // [b] assertEquals(1, provided.size()); - // [b, b, b, b, b] - assertEquals(5, processed.size()); + // [b, b, b, b] + assertEquals(4, processed.size()); // [] assertEquals(0, recovered.size()); assertEquals(1, stepExecution.getItemCount()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepHandlerTests.java new file mode 100644 index 000000000..5f7c0f4ac --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepHandlerTests.java @@ -0,0 +1,208 @@ +/* + * 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.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.step.handler.BasicAttributeAccessor; +import org.springframework.batch.core.step.item.SkipLimitStepFactoryBean.StatefulRetryStepHandler; +import org.springframework.batch.core.step.skip.ItemSkipPolicy; +import org.springframework.batch.core.step.skip.SkipLimitExceededException; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.NoWorkFoundException; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.retry.policy.NeverRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; + +/** + * @author Dave Syer + * + */ +public class StatefulRetryStepHandlerTests { + + private Log logger = LogFactory.getLog(getClass()); + + private int count = 0; + + private int limit = 3; + + protected int skipLimit = 2; + + protected List written = new ArrayList(); + + private StatefulRetryStepHandler handler; + + private RepeatTemplate chunkOperations = new RepeatTemplate(); + + private ItemReader itemReader = new ItemReader() { + public Integer read() { + return count++ >= limit ? null : count; + }; + }; + + private ItemWriter itemWriter = new ItemWriter() { + public void write(List items) throws Exception { + written.addAll(items); + } + }; + + private ItemProcessor itemProcessor = new ItemProcessor() { + public String process(Integer item) throws Exception { + return "" + item; + } + }; + + private RetryTemplate retryTemplate = new RetryTemplate(); + + private ItemSkipPolicy readSkipPolicy = new ItemSkipPolicy() { + public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException { + if (skipCount < skipLimit) { + return true; + } + throw new SkipLimitExceededException(skipLimit, t); + } + }; + + private ItemSkipPolicy writeSkipPolicy = readSkipPolicy; + + @Before + public void setUp() { + retryTemplate.setRetryPolicy(new NeverRetryPolicy()); + } + + @Test + public void testBasicHandle() throws Exception { + handler = new StatefulRetryStepHandler(itemReader, itemProcessor, itemWriter, chunkOperations, + retryTemplate, readSkipPolicy, writeSkipPolicy); + StepContribution contribution = new StepExecution("foo", null).createStepContribution(); + handler.handle(contribution, new BasicAttributeAccessor()); + assertEquals(limit, contribution.getItemCount()); + } + + @Test + public void testSkipOnRead() throws Exception { + handler = new StatefulRetryStepHandler(new ItemReader() { + public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { + throw new RuntimeException("Barf!"); + } + }, itemProcessor, itemWriter, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy); + chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); + StepContribution contribution = new StepExecution("foo", null).createStepContribution(); + BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + try { + handler.handle(contribution, attributes); + fail("Expected SkipLimitExceededException"); + } + catch (SkipLimitExceededException e) { + // expected + } + assertEquals(0, contribution.getItemCount()); + assertEquals(2, contribution.getReadSkipCount()); + } + + @Test + public void testSkipSingleItemOnWrite() throws Exception { + handler = new StatefulRetryStepHandler(itemReader, itemProcessor, new ItemWriter() { + public void write(List items) throws Exception { + written.addAll(items); + throw new RuntimeException("Barf!"); + } + }, chunkOperations, retryTemplate, readSkipPolicy, writeSkipPolicy); + chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); + StepContribution contribution = new StepExecution("foo", null).createStepContribution(); + BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + try { + handler.handle(contribution, attributes); + fail("Expected RuntimeException"); + } + catch (Exception e) { + assertEquals("Barf!", e.getMessage()); + } + assertTrue(attributes.hasAttribute("OUTPUT_BUFFER_KEY")); + handler.handle(contribution, attributes); + assertEquals(1, contribution.getItemCount()); + assertEquals(1, contribution.getWriteSkipCount()); + assertEquals(1, written.size()); + } + + @Test + public void testSkipMultipleItems() throws Exception { + handler = new StatefulRetryStepHandler(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.setCompletionPolicy(new SimpleCompletionPolicy(2)); + StepContribution contribution = new StepExecution("foo", null).createStepContribution(); + BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + + // Count to 3: (try + skip + skip) + for (int i = 0; i < 3; i++) { + try { + handler.handle(contribution, attributes); + fail("Expected RuntimeException on i="+i); + } + catch (Exception e) { + assertEquals("Barf!", e.getMessage()); + } + assertTrue(attributes.hasAttribute("OUTPUT_BUFFER_KEY")); + } + @SuppressWarnings("unchecked") + Chunk chunk = (Chunk) attributes.getAttribute("OUTPUT_BUFFER_KEY"); + assertEquals(1, chunk.getSkips().size()); + // The last recovery for this chunk... + handler.handle(contribution, attributes); + + attributes = new BasicAttributeAccessor(); + try { + handler.handle(contribution, attributes); + fail("Expected RuntimeException on i="); + } + catch (Exception e) { + assertEquals("Barf!", e.getMessage()); + } + try { + handler.handle(contribution, attributes); + fail("Expected SkipLimitExceededException"); + } + catch (SkipLimitExceededException e) { + // expected + } + assertTrue(attributes.hasAttribute("OUTPUT_BUFFER_KEY")); + assertEquals(3, contribution.getItemCount()); + assertEquals(2, contribution.getWriteSkipCount()); + assertEquals(5, written.size()); + } + +} diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/support/RetrySampleItemWriterTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/support/RetrySampleItemWriterTests.java index a1622675d..6b12d76cd 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/support/RetrySampleItemWriterTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/support/RetrySampleItemWriterTests.java @@ -26,7 +26,7 @@ public class RetrySampleItemWriterTests { processor.write(Collections.singletonList(item)); try { - processor.write(Arrays.asList(new Object[] { item, item, item })); + processor.write(Arrays.asList(item, item, item)); fail(); } catch (RuntimeException e) { @@ -35,6 +35,6 @@ public class RetrySampleItemWriterTests { processor.write(Collections.singletonList(item)); - assertEquals(4, processor.getCounter()); + assertEquals(5, processor.getCounter()); } }