BATCH-220: Implement skip on ItemProcessor execution

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<S> extends StepListener {
+public interface SkipListener<T,S> extends StepListener {
 
 	/**
 	 * Callback for a failure on read that is legal, so is not going to be
@@ -36,7 +36,7 @@ public interface SkipListener<S> 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<S> 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);
+
 }

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 <code>StepContribution</code>.
 	 */
 	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 + "]";
 	}
 
 }

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

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<S> implements SkipListener<S> {
+public class CompositeSkipListener<T,S> implements SkipListener<T,S> {
 
-	private OrderedComposite<SkipListener<? super S>> listeners = new OrderedComposite<SkipListener<? super S>>();
+	private OrderedComposite<SkipListener<? super T,? super S>> listeners = new OrderedComposite<SkipListener<? super T,? super S>>();
 
 	/**
 	 * Public setter for the listeners.
 	 * 
 	 * @param listeners
 	 */
-	public void setListeners(List<? extends SkipListener<? super S>> listeners) {
+	public void setListeners(List<? extends SkipListener<? super T,? super S>> listeners) {
 		this.listeners.setItems(listeners);
 	}
 
@@ -43,7 +43,7 @@ public class CompositeSkipListener<S> implements SkipListener<S> {
 	 * 
 	 * @param listener
 	 */
-	public void register(SkipListener<? super S> listener) {
+	public void register(SkipListener<? super T,? super S> listener) {
 		listeners.add(listener);
 	}
 
@@ -53,8 +53,8 @@ public class CompositeSkipListener<S> implements SkipListener<S> {
 	 * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable)
 	 */
 	public void onSkipInRead(Throwable t) {
-		for (Iterator<SkipListener<? super S>> iterator = listeners.iterator(); iterator.hasNext();) {
-			SkipListener<? super S> listener = iterator.next();
+		for (Iterator<SkipListener<? super T,? super S>> iterator = listeners.iterator(); iterator.hasNext();) {
+			SkipListener<? super T,? super S> listener = iterator.next();
 			listener.onSkipInRead(t);
 		}
 	}
@@ -66,9 +66,23 @@ public class CompositeSkipListener<S> implements SkipListener<S> {
 	 * java.lang.Throwable)
 	 */
 	public void onSkipInWrite(S item, Throwable t) {
-		for (Iterator<SkipListener<? super S>> iterator = listeners.iterator(); iterator.hasNext();) {
-			SkipListener<? super S> listener = iterator.next();
+		for (Iterator<SkipListener<? super T,? super S>> iterator = listeners.iterator(); iterator.hasNext();) {
+			SkipListener<? super T,? super S> 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<SkipListener<? super T,? super S>> iterator = listeners.iterator(); iterator.hasNext();) {
+			SkipListener<? super T,? super S> listener = iterator.next();
+			listener.onSkipInProcess(item, t);
+		}
+	}
+
 }

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<T, S> implements StepExecutionListener, ChunkListener, ItemReadListener<T>,
-		ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<S> {
+		ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<T,S> {
 
 	private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener();
 
@@ -45,7 +45,7 @@ public class MulticasterBatchListener<T, S> implements StepExecutionListener, Ch
 
 	private CompositeItemWriteListener<S> itemWriteListener = new CompositeItemWriteListener<S>();
 
-	private CompositeSkipListener<S> skipListener = new CompositeSkipListener<S>();
+	private CompositeSkipListener<T,S> skipListener = new CompositeSkipListener<T,S>();
 
 	/**
 	 * Initialise the listener instance.
@@ -298,4 +298,13 @@ public class MulticasterBatchListener<T, S> 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);
+	}
+
 }

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<T,S> implements SkipListener<T,S> {
 
 	/* (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) {		
+	}
 
 }

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

@@ -200,16 +200,19 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
 	 * @param contribution current context
 	 */
 	protected void process(StepContribution contribution, Chunk<T> inputs, Chunk<S> 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();
 	}
 

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

@@ -261,9 +261,8 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 			exceptions.addAll(new ArrayList<Class<? extends Throwable>>(retryableExceptionClasses));
 			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, readSkipPolicy,
-					writeSkipPolicy);
+			ChunkOrientedTasklet<T, S> tasklet = new StatefulRetryTasklet<T, S>(getItemReader(), getItemProcessor(),
+					getItemWriter(), getChunkOperations(), retryTemplate, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
 			tasklet.setListeners(getListeners());
 
 			step.setTasklet(tasklet);
@@ -304,6 +303,8 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 
 		final private ItemSkipPolicy writeSkipPolicy;
 
+		final private ItemSkipPolicy processSkipPolicy;
+
 		/**
 		 * @param itemReader
 		 * @param itemWriter
@@ -312,11 +313,12 @@ 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, 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<T, S> extends SimpleStepFactoryBean<T, S>
 
 		}
 
+		/**
+		 * 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<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();
+				
+				RetryCallback<S> retryCallback = new RetryCallback<S>() {
+
+					public S doWithRetry(RetryContext context) throws Exception {
+						contribution.incrementItemCount();
+						S output = doProcess(item);
+						return output;
+					}
+
+				};
+
+				RecoveryCallback<S> recoveryCallback = new RecoveryCallback<S>() {
+
+					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<T> 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.<br/>
 		 * 

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

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

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<Object,Object>() {
 			@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<Object,Object>() {
 			@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<Object,Object>() {
 			@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<Object,Object>() {
 			@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<Object,Object>() {
+			@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<Object,Object>() {
+			@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
 	 * 

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<String, String> handler = new ChunkOrientedTasklet<String, String>(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<String, String> handler = new ChunkOrientedTasklet<String, String>(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<String, String> {
+		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<? extends String> items) throws Exception {
 			for (String item : items) {
-				values += item;				
+				values += item;
 			}
 		}
 	}

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<String,String>() {
 			@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<String,String>() {
 			@Override
-			public void onSkipInWrite(Object item, Throwable t) {
+			public void onSkipInWrite(String item, Throwable t) {
 				throw new RuntimeException("oops");
 			}
 		} });

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<String,String>() {
+			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<String,String>() {
+			public void onSkipInWrite(String item, Throwable t) {
 				recovered.add(item);
 				assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
 			}

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<String> written = new ArrayList<String>();
+	private List<String> written = new ArrayList<String>();
 
+	private List<Integer> processed = new ArrayList<Integer>();
+	
 	private StatefulRetryTasklet<Integer, String> 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<Integer, String>(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<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, 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<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, 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<Integer> chunk = (Chunk<Integer>) 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());
+	}
 }