RESOLVED - issue BATCH-551: Put some checks in skip so that poor key generation algorithms do not cause weird looking failures.

Added skipCacheCapacity as configurable for skip limit factory bean.

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

@@ -25,6 +25,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.SkipListener;
 import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.UnexpectedJobExecutionException;
 import org.springframework.batch.core.listener.CompositeSkipListener;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
@@ -58,6 +59,8 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 
 	private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
 
+	private int skipCacheCapacity = 1024;
+
 	private Map skippedExceptions = new HashMap();
 
 	private ItemKeyGenerator defaultItemKeyGenerator = new ItemKeyGenerator() {
@@ -125,6 +128,20 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 		this.itemSkipPolicy = itemSkipPolicy;
 	}
 
+	/**
+	 * Public setter for the capacity of the skipped item cache. If a large
+	 * number of items are failing and not being recognized as skipped, it
+	 * usually signals a problem with the key generation (often equals and
+	 * hashCode in the item itself). So it is better to enforce a strict limit
+	 * than have weird looking errors, where a skip limit is reached without
+	 * anything being skipped.
+	 * 
+	 * @param skipCacheCapacity the capacity to set
+	 */
+	public void setSkipCacheCapacity(int skipCacheCapacity) {
+		this.skipCacheCapacity = skipCacheCapacity;
+	}
+
 	/**
 	 * Tries to read the item from the reader, in case of exception skip the
 	 * item if the skip policy allows, otherwise re-throw.
@@ -229,6 +246,13 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 	 */
 	private void addSkippedException(Object key, Throwable e) {
 		synchronized (skippedExceptions) {
+			if (skippedExceptions.size() >= skipCacheCapacity) {
+				throw new UnexpectedJobExecutionException(
+						"The cache of failed items to skipped unexpectedly reached its capacity ("
+								+ skipCacheCapacity
+								+ "). "
+								+ "This often indicates a problem with the key generation strategy, and/or a mistake in the implementation of hashCode and equals in the items being processed.");
+			}
 			skippedExceptions.put(key, e);
 		}
 	}

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

@@ -35,6 +35,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 
 	private ItemKeyGenerator itemKeyGenerator;
 
+	private int skipCacheCapacity = 1024;
+
 	/**
 	 * Public setter for a limit that determines skip policy. If this value is
 	 * positive then an exception in chunk processing will cause the item to be
@@ -86,7 +88,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 	public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
 		this.itemKeyGenerator = itemKeyGenerator;
 	}
-	
+
 	/**
 	 * Protected getter for the {@link ItemKeyGenerator}.
 	 * @return the {@link ItemKeyGenerator}
@@ -95,6 +97,24 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 		return itemKeyGenerator;
 	}
 
+	/**
+	 * Public setter for the capacity of the skipped item cache. If a large
+	 * number of items are failing and not being recognized as skipped, it
+	 * usually signals a problem with the key generation (often equals and
+	 * hashCode in the item itself). So it is better to enforce a strict limit
+	 * than have weird looking errors, where a skip limit is reached without
+	 * anything being skipped.<br/>
+	 * 
+	 * The default value is 1024 which should be high enough and more for most
+	 * purposes. To breach the limit in a single-threaded step typically you
+	 * have to have this many failures in a single transaction.
+	 * 
+	 * @param skipCacheCapacity the capacity to set
+	 */
+	public void setSkipCacheCapacity(int skipCacheCapacity) {
+		this.skipCacheCapacity = skipCacheCapacity;
+	}
+
 	/**
 	 * Uses the {@link #skipLimit} value to configure item handler and and
 	 * exception handler.
@@ -113,7 +133,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			 */
 			addFatalExceptionIfMissing(SkipLimitExceededException.class);
 			List fatalExceptionList = Arrays.asList(fatalExceptionClasses);
-			
+
 			LimitCheckingItemSkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, Arrays
 					.asList(skippableExceptionClasses), fatalExceptionList);
 			itemHandler.setItemSkipPolicy(skipPolicy);
@@ -127,6 +147,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			setExceptionHandler(exceptionHandler);
 
 			itemHandler.setItemKeyGenerator(itemKeyGenerator);
+			itemHandler.setSkipCacheCapacity(skipCacheCapacity);
 
 			BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
 			itemHandler.setSkipListeners(helper.getSkipListeners(getListeners()));