BATCH-798: Fixed bug in skip on process (the wrong state was being used for retry).

spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java

@@ -57,6 +57,13 @@ public class StepContribution {
 		itemCount++;
 	}
 
+	/**
+	 * Increment the counter for the number of items processed.
+	 */
+	public void incrementItemCount(int count) {
+		itemCount+=count;
+	}
+
 	/**
 	 * Public access to the item counter.
 	 * 

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

@@ -19,7 +19,7 @@ class Chunk<W> implements Iterable<W> {
 
 	private List<W> items = new ArrayList<W>();
 
-	private List<SkippedItem<W>> skips = new ArrayList<SkippedItem<W>>();
+	private List<ItemWrapper<W>> skips = new ArrayList<ItemWrapper<W>>();
 
 	/**
 	 * Add the item to the chunk.
@@ -46,8 +46,8 @@ class Chunk<W> implements Iterable<W> {
 	/**
 	 * @return a copy of the skips as an unmodifiable list
 	 */
-	public List<SkippedItem<W>> getSkips() {
-		return Collections.unmodifiableList(new ArrayList<SkippedItem<W>>(skips));
+	public List<ItemWrapper<W>> getSkips() {
+		return Collections.unmodifiableList(new ArrayList<ItemWrapper<W>>(skips));
 	}
 
 	/**
@@ -117,7 +117,7 @@ class Chunk<W> implements Iterable<W> {
 					return;
 				}
 			}
-			skips.add(new SkippedItem<W>(next, e));
+			skips.add(new ItemWrapper<W>(next, e));
 			iterator.remove();
 		}
 
@@ -127,44 +127,4 @@ class Chunk<W> implements Iterable<W> {
 
 	}
 
-	/**
-	 * Wrapper for a skipped item and its exception.
-	 * 
-	 * @author Dave Syer
-	 * 
-	 */
-	public static class SkippedItem<T> {
-
-		final private Exception exception;
-
-		final private T item;
-
-		public SkippedItem(T item, Exception e) {
-			this.item = item;
-			this.exception = e;
-		}
-
-		/**
-		 * Public getter for the exception.
-		 * @return the exception
-		 */
-		public Exception getException() {
-			return exception;
-		}
-
-		/**
-		 * Public getter for the item.
-		 * @return the item
-		 */
-		public T getItem() {
-			return item;
-		}
-
-		@Override
-		public String toString() {
-			return String.format("[exception=%s, item=%s]", exception, item);
-		}
-
-	}
-
 }

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

@@ -120,7 +120,7 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 	public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception {
 
 		// TODO: check flags to see if these need to be saved or not (e.g. JMS not)
-		final Chunk<T> inputs = getInputBuffer(attributes);
+		final Chunk<ItemWrapper<T>> inputs = getInputBuffer(attributes);
 		final Chunk<S> outputs = getOutputBuffer(attributes);
 
 		ExitStatus result = ExitStatus.CONTINUABLE;
@@ -134,7 +134,7 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 					if (item.getItem() == null) {
 						return ExitStatus.FINISHED;
 					}
-					inputs.add(item.getItem());
+					inputs.add(item);
 					return ExitStatus.CONTINUABLE;
 				}
 			});
@@ -199,9 +199,9 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 	 * @param outputs the items to write
 	 * @param contribution current context
 	 */
-	protected void process(StepContribution contribution, Chunk<T> inputs, Chunk<S> outputs) throws Exception {
+	protected void process(StepContribution contribution, Chunk<ItemWrapper<T>> inputs, Chunk<S> outputs) throws Exception {
 		int filtered = 0;
-		for (T item : inputs) {
+		for (ItemWrapper<T> item : inputs) {
 			// TODO: segregate read / write / filter count
 			// (this is read count)
 			contribution.incrementItemCount();
@@ -217,11 +217,12 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 	}
 
 	/**
-	 * @param item the input item
+	 * @param wrapper the input item
 	 * @return the result of the processing
 	 * @throws Exception
 	 */
-	protected S doProcess(T item) throws Exception {
+	protected S doProcess(ItemWrapper<T> wrapper) throws Exception {
+		T item = wrapper.getItem();
 		try {
 			listener.beforeProcess(item);
 			S result = itemProcessor.process(item);
@@ -272,7 +273,7 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 	 * @param attributes
 	 * @param inputs
 	 */
-	private void storeInputs(AttributeAccessor attributes, Chunk<T> inputs) {
+	private void storeInputs(AttributeAccessor attributes, Chunk<ItemWrapper<T>> inputs) {
 		store(attributes, INPUT_BUFFER_KEY, inputs);
 	}
 
@@ -306,7 +307,7 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 		}
 	}
 
-	private Chunk<T> getInputBuffer(AttributeAccessor attributes) {
+	private Chunk<ItemWrapper<T>> getInputBuffer(AttributeAccessor attributes) {
 		return getBuffer(attributes, INPUT_BUFFER_KEY);
 	}
 
@@ -328,57 +329,4 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 		return resource;
 	}
 
-	/**
-	 * @author Dave Syer
-	 * 
-	 */
-	static protected class ItemWrapper<T> {
-
-		final private T item;
-
-		final private int skipCount;
-
-		/**
-		 * @param item
-		 */
-		public ItemWrapper(T item) {
-			this(item, 0);
-		}
-
-		/**
-		 * @param item
-		 * @param skipCount
-		 */
-		public ItemWrapper(T item, int skipCount) {
-			this.item = item;
-			this.skipCount = skipCount;
-		}
-
-		/**
-		 * @return the item we are wrapping
-		 */
-		public T getItem() {
-			return item;
-		}
-
-		/**
-		 * Public getter for the skipCount.
-		 * @return the skipCount
-		 */
-		public int getSkipCount() {
-			return skipCount;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * 
-		 * @see java.lang.Object#toString()
-		 */
-		@Override
-		public String toString() {
-			return String.format("[%s,%d]", item, skipCount);
-		}
-
-	}
-
 }

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

@@ -0,0 +1,71 @@
+package org.springframework.batch.core.step.item;
+
+/**
+ * Wrapper for an item and its exception if it failed processing.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class ItemWrapper<T> {
+
+	final private Exception exception;
+
+	final private T item;
+
+	final private int skipCount;
+
+	/**
+	 * @param item
+	 */
+	public ItemWrapper(T item) {
+		this(item, null, 0);
+	}
+
+	/**
+	 * @param item
+	 * @param skipCount
+	 */
+	public ItemWrapper(T item, int skipCount) {
+		this(item, null, skipCount);
+	}
+
+	public ItemWrapper(T item, Exception e) {
+		this(item, e, 0);
+	}
+
+	public ItemWrapper(T item, Exception e, int skipCount) {
+		this.item = item;
+		this.exception = e;
+		this.skipCount = skipCount;
+	}
+
+	/**
+	 * Public getter for the skipCount.
+	 * @return the skipCount
+	 */
+	public int getSkipCount() {
+		return skipCount;
+	}
+
+	/**
+	 * Public getter for the exception.
+	 * @return the exception
+	 */
+	public Exception getException() {
+		return exception;
+	}
+
+	/**
+	 * Public getter for the item.
+	 * @return the item
+	 */
+	public T getItem() {
+		return item;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("[exception=%s, item=%s, skips=%d]", exception, item, skipCount);
+	}
+
+}

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

@@ -261,7 +261,8 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 			ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
 					new ArrayList<Class<? extends Throwable>>(fatalExceptionClasses));
 			ChunkOrientedTasklet<T, S> tasklet = new StatefulRetryTasklet<T, S>(getItemReader(), getItemProcessor(),
-					getItemWriter(), getChunkOperations(), retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
+					getItemWriter(), getChunkOperations(), retryTemplate, rollbackClassifier, readSkipPolicy,
+					writeSkipPolicy, writeSkipPolicy);
 			tasklet.setListeners(getListeners());
 
 			step.setTasklet(tasklet);
@@ -313,7 +314,8 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 		 */
 		public StatefulRetryTasklet(ItemReader<? extends T> itemReader,
 				ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
-				RepeatOperations chunkOperations, RetryOperations retryTemplate, Classifier<Throwable, Boolean> rollbackClassifier, ItemSkipPolicy readSkipPolicy,
+				RepeatOperations chunkOperations, RetryOperations retryTemplate,
+				Classifier<Throwable, Boolean> rollbackClassifier, ItemSkipPolicy readSkipPolicy,
 				ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) {
 			super(itemReader, itemProcessor, itemWriter, chunkOperations);
 			this.retryOperations = retryTemplate;
@@ -378,20 +380,19 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 		 * org.springframework.batch.core.step.item.Chunk)
 		 */
 		@Override
-		protected void process(final StepContribution contribution, final Chunk<T> inputs, final Chunk<S> outputs)
+		protected void process(final StepContribution contribution, final Chunk<ItemWrapper<T>> inputs, final Chunk<S> outputs)
 				throws Exception {
 
 			int filtered = 0;
 
-			for (final Chunk<T>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
-				
-				final T item = iterator.next();
-				
+			for (final Chunk<ItemWrapper<T>>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
+
+				final ItemWrapper<T> wrapper = iterator.next();
+
 				RetryCallback<S> retryCallback = new RetryCallback<S>() {
 
 					public S doWithRetry(RetryContext context) throws Exception {
-						contribution.incrementItemCount();
-						S output = doProcess(item);
+						S output = doProcess(wrapper);
 						return output;
 					}
 
@@ -400,43 +401,40 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 				RecoveryCallback<S> recoveryCallback = new RecoveryCallback<S>() {
 
 					public S recover(RetryContext context) throws Exception {
-						Exception e = context.getLastThrowable();
+						Exception e = (Exception) context.getLastThrowable();
 						if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
 							contribution.incrementProcessSkipCount();
 							iterator.remove(e);
+							return null;
 						}
 						else {
-							throw new RetryException("Non-skippable exception in recoverer", e);
+							throw new RetryException("Non-skippable exception in recoverer while processing", e);
 						}
-						// Unless we reached the end of the chunk we need to rethrow
-						if (iterator.hasNext()) {
-							throw e;
-						}
-						return null;
 					}
-					
+
 				};
 
-				S output = retryOperations.execute(retryCallback, recoveryCallback, new RetryState(inputs));
-				// TODO: increment filter count if this is null
+				S output = retryOperations.execute(retryCallback, recoveryCallback, new RetryState(wrapper));
 				if (output != null) {
 					outputs.add(output);
-				} else {
+				}
+				else {
 					filtered++;
 				}
 
 			}
 
-			for (Chunk.SkippedItem<T> skip : inputs.getSkips()) {
+			for (ItemWrapper<ItemWrapper<T>> skip : inputs.getSkips()) {
 				Exception exception = skip.getException();
 				try {
-					getListener().onSkipInProcess(skip.getItem(), exception);
+					getListener().onSkipInProcess(skip.getItem().getItem(), exception);
 				}
 				catch (RuntimeException e) {
 					throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception);
 				}
 			}
-			
+
+			contribution.incrementItemCount(inputs.size());
 			contribution.incrementFilterCount(filtered);
 
 			inputs.clear();
@@ -457,8 +455,6 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 			RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
 				public Object doWithRetry(RetryContext context) throws Exception {
 					doWrite(chunk.getItems());
-					// TODO: if there is an exception marked as no rollback it
-					// should get treated as stateless
 					return null;
 				}
 			};
@@ -506,7 +502,7 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 
 			retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk));
 
-			for (Chunk.SkippedItem<S> skip : chunk.getSkips()) {
+			for (ItemWrapper<S> skip : chunk.getSkips()) {
 				Exception exception = skip.getException();
 				try {
 					getListener().onSkipInWrite(skip.getItem(), exception);

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

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.core.step.item;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class ItemWrapperTests {
+
+	private Exception exception = new RuntimeException();
+
+	/**
+	 * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object)}.
+	 */
+	@Test
+	public void testItemWrapperT() {
+		ItemWrapper<String> wrapper = new ItemWrapper<String>("foo");
+		assertEquals("foo", wrapper.getItem());
+		assertEquals(null, wrapper.getException());
+		assertEquals(0, wrapper.getSkipCount());
+	}
+
+	/**
+	 * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, int)}.
+	 */
+	@Test
+	public void testItemWrapperTInt() {
+		ItemWrapper<String> wrapper = new ItemWrapper<String>("foo",2);
+		assertEquals("foo", wrapper.getItem());
+		assertEquals(null, wrapper.getException());
+		assertEquals(2, wrapper.getSkipCount());
+	}
+
+	/**
+	 * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, java.lang.Exception)}.
+	 */
+	@Test
+	public void testItemWrapperTException() {
+		ItemWrapper<String> wrapper = new ItemWrapper<String>("foo",exception);
+		assertEquals("foo", wrapper.getItem());
+		assertEquals(exception, wrapper.getException());
+		assertEquals(0, wrapper.getSkipCount());
+	}
+
+	/**
+	 * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, java.lang.Exception, int)}.
+	 */
+	@Test
+	public void testItemWrapperTExceptionInt() {
+		ItemWrapper<String> wrapper = new ItemWrapper<String>("foo", exception, 2);
+		assertEquals("foo", wrapper.getItem());
+		assertEquals(exception , wrapper.getException());
+		assertEquals(2, wrapper.getSkipCount());
+	}
+
+	/**
+	 * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#toString()}.
+	 */
+	@Test
+	public void testToString() {
+		ItemWrapper<String> wrapper = new ItemWrapper<String>("foo");
+		assertTrue("foo", wrapper.toString().contains("foo"));
+	}
+
+}

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

@@ -49,7 +49,7 @@ import org.springframework.batch.support.Classifier;
  * 
  */
 public class StatefulRetryTaskletTests {
-	
+
 	private Log logger = LogFactory.getLog(getClass());
 
 	private int count = 0;
@@ -61,7 +61,7 @@ public class StatefulRetryTaskletTests {
 	private List<String> written = new ArrayList<String>();
 
 	private List<Integer> processed = new ArrayList<Integer>();
-	
+
 	private StatefulRetryTasklet<Integer, String> handler;
 
 	private RepeatTemplate chunkOperations = new RepeatTemplate();
@@ -85,7 +85,7 @@ public class StatefulRetryTaskletTests {
 	};
 
 	private RetryTemplate retryTemplate = new RetryTemplate();
-	
+
 	private Classifier<Throwable, Boolean> rollbackClassifier = new Classifier<Throwable, Boolean>() {
 		public Boolean classify(Throwable classifiable) {
 			return true;
@@ -103,7 +103,6 @@ public class StatefulRetryTaskletTests {
 
 	private ItemSkipPolicy writeSkipPolicy = readSkipPolicy;
 
-
 	@Before
 	public void setUp() {
 		retryTemplate.setRetryPolicy(new NeverRetryPolicy());
@@ -124,7 +123,8 @@ public class StatefulRetryTaskletTests {
 			public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
 				throw new RuntimeException("Barf!");
 			}
-		}, itemProcessor, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
+		}, itemProcessor, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy,
+				writeSkipPolicy, writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
 		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
@@ -168,7 +168,7 @@ public class StatefulRetryTaskletTests {
 	public void testSkipMultipleItemsOnWrite() throws Exception {
 		handler = new StatefulRetryTasklet<Integer, String>(itemReader, itemProcessor, new ItemWriter<String>() {
 			public void write(List<? extends String> items) throws Exception {
-				logger.debug("Writing items: "+items);
+				logger.debug("Writing items: " + items);
 				written.addAll(items);
 				throw new RuntimeException("Barf!");
 			}
@@ -176,12 +176,12 @@ public class StatefulRetryTaskletTests {
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
 		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
-		
-		// Count to 3: (try + skip + skip) 
+
+		// Count to 3: (try + skip + skip)
 		for (int i = 0; i < 3; i++) {
 			try {
 				handler.execute(contribution, attributes);
-				fail("Expected RuntimeException on i="+i);
+				fail("Expected RuntimeException on i=" + i);
 			}
 			catch (Exception e) {
 				assertEquals("Barf!", e.getMessage());
@@ -216,24 +216,64 @@ public class StatefulRetryTaskletTests {
 	}
 
 	@Test
-	public void testSkipMultipleItemsOnProcess() throws Exception {
+	public void testSkipSingleItemOnProcess() throws Exception {
 		handler = new StatefulRetryTasklet<Integer, String>(itemReader, new ItemProcessor<Integer, String>() {
 			public String process(Integer item) throws Exception {
-				logger.debug("Processing item: "+item);
+				logger.debug("Processing item: " + item);
+				processed.add(item);
+				if (item == 3) {
+					throw new RuntimeException("Barf!");
+				}
+				return "p" + item;
+			}
+		}, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy,
+				writeSkipPolicy);
+		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3));
+		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
+		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
+
+		// try
+		try {
+			handler.execute(contribution, attributes);
+			fail("Expected RuntimeException");
+		}
+		catch (Exception e) {
+			assertEquals("Barf!", e.getMessage());
+		}
+		assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY"));
+
+		@SuppressWarnings("unchecked")
+		Chunk<Integer> chunk = (Chunk<Integer>) attributes.getAttribute("INPUT_BUFFER_KEY");
+
+		// skip...
+		handler.execute(contribution, attributes);
+		assertEquals(1, chunk.getSkips().size());
+
+		assertEquals(2, contribution.getItemCount());
+		assertEquals(1, contribution.getProcessSkipCount());
+		assertEquals(5, processed.size());
+		assertEquals("[p1, p2]", written.toString());
+	}
+
+	@Test
+	public void testSkipOverLimitOnProcess() throws Exception {
+		handler = new StatefulRetryTasklet<Integer, String>(itemReader, new ItemProcessor<Integer, String>() {
+			public String process(Integer item) throws Exception {
+				logger.debug("Processing item: " + item);
 				processed.add(item);
 				throw new RuntimeException("Barf!");
 			}
-		}
-		, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
+		}, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy,
+				writeSkipPolicy);
 		chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2));
 		StepContribution contribution = new StepExecution("foo", null).createStepContribution();
 		BasicAttributeAccessor attributes = new BasicAttributeAccessor();
-		
-		// Count to 3: (try + skip + try) 
-		for (int i = 0; i < 3; i++) {
+
+		// Count to 3: (try + skip + try)
+		for (int i = 0; i < 2; i++) {
 			try {
 				handler.execute(contribution, attributes);
-				fail("Expected RuntimeException on i="+i);
+				fail("Expected RuntimeException on i=" + i);
 			}
 			catch (Exception e) {
 				assertEquals("Barf!", e.getMessage());
@@ -264,8 +304,9 @@ public class StatefulRetryTaskletTests {
 			// expected
 		}
 		assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY"));
-		assertEquals(3, contribution.getItemCount());
+		assertEquals(0, contribution.getItemCount());
 		assertEquals(2, contribution.getProcessSkipCount());
+		// Just before the skip at the end we process once more
 		assertEquals(3, processed.size());
 	}
 }