OPEN - issue BATCH-220: Chunk-oriented approach to processing

Add small optimisation to skip chunk with single item

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<T, S> extends StepListener {
+
+	void beforeProcess(T item);
+	
+	void afterProcess(T item, S result);
+	
+	void onProcessError(T item, Exception e);
+}

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<T, S> implements ItemProcessListener<T, S> {
+
+	private OrderedComposite<ItemProcessListener<? super T, ? super S>> listeners = new OrderedComposite<ItemProcessListener<? super T, ? super S>>();
+
+	/**
+	 * Public setter for the listeners.
+	 * 
+	 * @param itemReadListeners
+	 */
+	public void setListeners(List<? extends ItemProcessListener<? super T, ? super S>> itemReadListeners) {
+		this.listeners.setItems(itemReadListeners);
+	}
+
+	/**
+	 * Register additional listener.
+	 * 
+	 * @param itemReaderListener
+	 */
+	public void register(ItemProcessListener<? super T, ? super S> 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<ItemProcessListener<? super T, ? super S>> iterator = listeners.reverse(); iterator.hasNext();) {
+			ItemProcessListener<? super T, ? super S> 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<ItemProcessListener<? super T, ? super S>> iterator = listeners.iterator(); iterator.hasNext();) {
+			ItemProcessListener<? super T, ? super S> 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<ItemProcessListener<? super T, ? super S>> iterator = listeners.reverse(); iterator.hasNext();) {
+			ItemProcessListener<? super T, ? super S> listener = iterator.next();
+			listener.onProcessError(item, e);
+		}
+	}
+
+}

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<T, S> implements StepExecutionListener, ChunkListener, ItemReadListener<T>,
-		ItemWriteListener<S>, SkipListener<S> {
+		ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<S> {
 
 	private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener();
 
@@ -40,6 +41,8 @@ public class MulticasterBatchListener<T, S> implements StepExecutionListener, Ch
 
 	private CompositeItemReadListener<T> itemReadListener = new CompositeItemReadListener<T>();
 
+	private CompositeItemProcessListener<T, S> itemProcessListener = new CompositeItemProcessListener<T, S>();
+
 	private CompositeItemWriteListener<S> itemWriteListener = new CompositeItemWriteListener<S>();
 
 	private CompositeSkipListener<S> skipListener = new CompositeSkipListener<S>();
@@ -76,10 +79,13 @@ public class MulticasterBatchListener<T, S> 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<T, S> 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)
 	 */

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 {
+}

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;

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java

@@ -65,6 +65,13 @@ class Chunk<W> implements Iterable<W> {
 		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<W> implements Iterable<W> {
 		}
 
 		public void remove(Exception e) {
-			if (next != null) {
-				skips.add(new SkippedItem<W>(next, e));
+			if (next == null) {
+				if (iterator.hasNext()) {
+					next = iterator.next();
+				}
+				else {
+					return;
+				}
 			}
+			skips.add(new SkippedItem<W>(next, e));
 			iterator.remove();
 		}
 

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java

@@ -78,6 +78,36 @@ public class ItemOrientedStepHandler<T, S> 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<T,S> 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<T, S> 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<T, S> implements StepHandler {
 
 	}
 
+	/**
+	 * @param contribution current context
+	 * @return next item for writing
+	 */
+	protected ItemWrapper<T> read(StepContribution contribution) throws Exception {
+		return new ItemWrapper<T>(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<T> inputs, Chunk<S> 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<S> chunk, StepContribution contribution) throws Exception {
+		doWrite(chunk.getItems());
+		chunk.clear();
+	}
+
+	/**
+	 * @param items
+	 * @throws Exception
+	 */
+	protected final void doWrite(List<S> 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<T, S> implements StepHandler {
 		return resource;
 	}
 
-	/**
-	 * @param contribution current context
-	 * @return next item for writing
-	 */
-	protected ItemWrapper<T> read(StepContribution contribution) throws Exception {
-		return new ItemWrapper<T>(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<S> chunk, StepContribution contribution) throws Exception {
-		doWrite(chunk.getItems());
-		chunk.clear();
-	}
-
-	/**
-	 * @param items
-	 * @throws Exception
-	 */
-	protected final void doWrite(List<S> 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<T,S> getListener() {
-		return listener;
-	}
-
 	/**
 	 * @author Dave Syer
 	 * 

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java

@@ -390,7 +390,13 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 
 				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<S>.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) {
 						S item = iterator.next();
@@ -398,20 +404,25 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 							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<S>.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));

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<Object, Object> listener;
+
+	private CompositeItemProcessListener<Object, Object> compositeListener;
+
+	@SuppressWarnings("unchecked")
+	@Before
+	public void setUp() throws Exception {
+		listener = createMock(ItemProcessListener.class);
+		compositeListener = new CompositeItemProcessListener<Object, Object>();
+		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<ItemProcessListener<? super Object, ? super Object>>() {
+			{
+				add(listener);
+			}
+		});
+		listener.beforeProcess(null);
+		replay(listener);
+		compositeListener.beforeProcess(null);
+		verify(listener);
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java

@@ -73,7 +73,7 @@ public class StatefulRetryStepFactoryBeanTests {
 	private List<Object> provided = new ArrayList<Object>();
 
 	private List<Object> 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());

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<String> written = new ArrayList<String>();
+
+	private StatefulRetryStepHandler<Integer, String> handler;
+
+	private RepeatTemplate chunkOperations = new RepeatTemplate();
+
+	private ItemReader<Integer> itemReader = new ItemReader<Integer>() {
+		public Integer read() {
+			return count++ >= limit ? null : count;
+		};
+	};
+
+	private ItemWriter<String> itemWriter = new ItemWriter<String>() {
+		public void write(List<? extends String> items) throws Exception {
+			written.addAll(items);
+		}
+	};
+
+	private ItemProcessor<Integer, String> itemProcessor = new ItemProcessor<Integer, String>() {
+		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<Integer, String>(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<Integer, String>(new ItemReader<Integer>() {
+			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<Integer, String>(itemReader, itemProcessor, new ItemWriter<String>() {
+			public void write(List<? extends String> 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<Integer, String>(itemReader, itemProcessor, new ItemWriter<String>() {
+			public void write(List<? extends String> 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<String> chunk = (Chunk<String>) 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());
+	}
+
+}

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());
 	}
 }