REOPENED - BATCH-896: "DRY" FaultTolerantTasklet implementations

added skip buffer cleaning and call skip listeners cautiously

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

@@ -1,11 +1,14 @@
 package org.springframework.batch.core.step.item;
 
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
+import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
@@ -70,24 +73,74 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends
 		// filter inputs marked for skipping
 		final Map<I, Exception> skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY);
 		final Map<O, Exception> skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY);
+		final Set<I> inputsIncludingSkips = new HashSet<I>(inputs.size());
+		final Set<O> outputsIncludingSkips = new HashSet<O>(inputs.size());
 
 		if (!inputs.isEmpty()) {
+			inputsIncludingSkips.addAll(inputs);
 			inputs.removeAll(skippedInputs.keySet());
 
 			final List<O> outputs = new ArrayList<O>();
 			process(contribution, inputs, outputs, skippedInputs);
 
 			// filter outputs marked for skipping
+			outputsIncludingSkips.addAll(outputs);
 			outputs.removeAll(skippedOutputs.keySet());
 
 			write(outputs, contribution, skippedOutputs);
 		}
 
-		callSkipListeners(skippedReads, skippedInputs, skippedOutputs);
+		callSkipListenersAndCleanSkipsFromBuffer(skippedReads, skippedInputs, skippedOutputs, inputsIncludingSkips,
+				outputsIncludingSkips);
 
 		return result;
 	}
 
+	/**
+	 * Identify items successfully skipped in this tasklet iteration, call skip
+	 * listeners and remove skips from buffer. This requires care, because we
+	 * might be processing a different chunk after rollback i.e. items marked
+	 * for skipping from previous tasklet iteration may not have been
+	 * encountered now.
+	 */
+	private void callSkipListenersAndCleanSkipsFromBuffer(final List<Exception> skippedReads,
+			final Map<I, Exception> skippedInputs, final Map<O, Exception> skippedOutputs,
+			final Set<I> inputsIncludingSkips, final Set<O> outputsIncludingSkips) {
+		for (Exception skippedReadException : skippedReads) {
+			try {
+				listener.onSkipInRead(skippedReadException);
+			}
+			catch (RuntimeException e) {
+				throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, skippedReadException);
+			}
+		}
+		skippedReads.clear();
+		for (I input : inputsIncludingSkips) {
+			if (skippedInputs.containsKey(input)) {
+				try {
+					listener.onSkipInProcess(input, skippedInputs.get(input));
+				}
+				catch (RuntimeException ex) {
+					throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skippedInputs
+							.get(input));
+				}
+				skippedInputs.remove(input);
+			}
+		}
+		for (O output : outputsIncludingSkips) {
+			if (skippedOutputs.containsKey(output)) {
+				try {
+					listener.onSkipInWrite(output, skippedOutputs.get(output));
+				}
+				catch (RuntimeException ex) {
+					throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skippedOutputs
+							.get(output));
+				}
+				skippedOutputs.remove(output);
+			}
+		}
+	}
+
 	/**
 	 * Tries to read the item from the reader, in case of exception skip the
 	 * skip listener is called and exception is re-thrown (failed read causes