OPEN - issue BATCH-498: Skip method should get ExecutionContext as an argument

Handle skip internally in ItemHandler.

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

@@ -15,12 +15,17 @@
  */
 package org.springframework.batch.core.step.item;
 
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
+import org.springframework.batch.item.ItemKeyGenerator;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.item.Skippable;
 
 /**
  * {@link ItemHandler} that implements skip behavior. It delegates to
@@ -35,8 +40,32 @@ import org.springframework.batch.item.Skippable;
  */
 public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 
+	protected final Log logger = LogFactory.getLog(getClass());
+
 	private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
 
+	private ItemKeyGenerator itemKeyGenerator = new ItemKeyGenerator() {
+		public Object getKey(Object item) {
+			return item;
+		}
+	};
+
+	private Set skippedItems = new HashSet();
+
+	/**
+	 * Public setter for the {@link ItemKeyGenerator}. Defaults to just return
+	 * the item, and since it will be used before a write operation.
+	 * Implementations must ensure that items always have the same key when they
+	 * are read from the {@link ItemReader} (so if the item is mutable and the
+	 * reader does any buffering the key generator might need to take care to
+	 * only use data that do not change on write).
+	 * 
+	 * @param itemKeyGenerator the itemKeyGenerator to set
+	 */
+	public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
+		this.itemKeyGenerator = itemKeyGenerator;
+	}
+
 	/**
 	 * @param itemReader
 	 * @param itemWriter
@@ -53,50 +82,62 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 	}
 
 	/**
-	 * Tries to read the item from the reader, in case exception is thrown calls
-	 * skip on the reader (if skipPolicy decides it is appropriate) before
-	 * rethrowing the exception.
+	 * Tries to read the item from the reader, in case of exception skip the
+	 * item if the skip policy allows, otherwise re-throw.
 	 * 
 	 * @param contribution current StepContribution holding skipped items count
 	 * @return next item for processing
 	 */
 	protected Object read(StepContribution contribution) throws Exception {
-		try {
-			return getItemReader().read();
-		}
-		catch (Exception e) {
-			if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
-				contribution.incrementSkipCount();
-				if (getItemReader() instanceof Skippable) {
-					((Skippable) getItemReader()).skip();
+
+		while (true) {
+
+			try {
+
+				Object item = doRead();
+				while (item != null && skippedItems.contains(itemKeyGenerator.getKey(item))) {
+					logger.debug("Skipping item on input: " + item);
+					item = doRead();
+				}
+				return item;
+
+			}
+			catch (Exception e) {
+				if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+					// increment skip count and try again
+					contribution.incrementSkipCount();
+				}
+				else {
+					// re-throw only when the skip policy runs out of patience
+					throw e;
 				}
 			}
-			throw e;
+
 		}
+
 	}
 
 	/**
-	 * Tries to write the item using the writer, in case exception is thrown
-	 * calls skip on both reader and writer (if skipPolicy decides it is
-	 * appropriate) before rethrowing the exception.
+	 * Tries to write the item using the writer. In case of exception consults
+	 * skip policy before re-throwing the exception. The exception is always
+	 * re-thrown, but if the item is seen again on read it will be skipped.
 	 * 
 	 * @param item item to write
 	 * @param contribution current StepContribution holding skipped items count
 	 */
 	protected void write(Object item, StepContribution contribution) throws Exception {
+		// Get the key as early as possible, otherwise it might change in
+		// doWrite()
+		Object key = itemKeyGenerator.getKey(item);
 		try {
-			getItemWriter().write(item);
+			doWrite(item);
 		}
 		catch (Exception e) {
 			if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
 				contribution.incrementSkipCount();
-				if (getItemReader() instanceof Skippable) {
-					((Skippable) getItemReader()).skip();
-				}
-				if (getItemWriter() instanceof Skippable) {
-					((Skippable) getItemWriter()).skip();
-				}
+				skippedItems.add(key);
 			}
+			// always re-throw exception on write
 			throw e;
 		}
 	}

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

@@ -52,22 +52,6 @@ public class SimpleItemHandler implements ItemHandler {
 		this.itemWriter = itemWriter;
 	}
 
-	/**
-	 * Public getter for the ItemReader.
-	 * @return the itemReader
-	 */
-	public ItemReader getItemReader() {
-		return itemReader;
-	}
-
-	/**
-	 * Public getter for the ItemWriter.
-	 * @return the itemWriter
-	 */
-	public ItemWriter getItemWriter() {
-		return itemWriter;
-	}
-
 	/**
 	 * Get the next item from {@link #read(StepContribution)} and if not null
 	 * pass the item to {@link #write(Object, StepContribution)}.
@@ -88,6 +72,14 @@ public class SimpleItemHandler implements ItemHandler {
 	 * @return next item for writing
 	 */
 	protected Object read(StepContribution contribution) throws Exception {
+		return doRead();
+	}
+
+	/**
+	 * @return
+	 * @throws Exception
+	 */
+	protected final Object doRead() throws Exception {
 		return itemReader.read();
 	}
 
@@ -97,6 +89,14 @@ public class SimpleItemHandler implements ItemHandler {
 	 * @param contribution current context
 	 */
 	protected void write(Object item, StepContribution contribution) throws Exception {
+		doWrite(item);
+	}
+
+	/**
+	 * @param item
+	 * @throws Exception
+	 */
+	protected final void doWrite(Object item) throws Exception {
 		itemWriter.write(item);
 	}