diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
index f7e7eb9cb..dbcf633ae 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
+++ b/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);
}
}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
index 463c739c2..8b7a90382 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
+++ b/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.
+ *
+ * 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()));
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java
index babace9df..c64fc35c9 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java
@@ -26,6 +26,7 @@ import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.job.JobSupport;
import org.springframework.batch.core.step.StepSupport;
import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
@@ -184,6 +185,29 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase {
assertEquals(new Holder("5"), handler.read(contribution));
}
+ public void testWriteWithSkipCapacitBreached() throws Exception {
+ handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
+ handler.setSkipCacheCapacity(0);
+ handler.handle(contribution);
+ handler.handle(contribution);
+ contribution.combineSkipCounts();
+ assertEquals(1, contribution.getSkipCount());
+ // 2 is skipped so 3 was last one processed and now we are at 4
+ try {
+ handler.handle(contribution);
+ fail("Expected UnexpectedJobExecutionException");
+ }
+ catch (UnexpectedJobExecutionException e) {
+ // expected
+ String message = e.getMessage();
+ assertTrue("Message does not contain 'capacity': "+message, message.indexOf("capacity")>=0);
+ }
+ assertEquals(2, contribution.getSkipCount());
+ // No "4" because it was skipped on write, even though it is mutating
+ // its key
+ assertEquals(new Holder("5"), handler.read(contribution));
+ }
+
/**
* Simple item reader that supports skip functionality.
*/
diff --git a/spring-batch-samples/.springBeans b/spring-batch-samples/.springBeans
index 602a737ec..5fb0921f1 100644
--- a/spring-batch-samples/.springBeans
+++ b/spring-batch-samples/.springBeans
@@ -1,10 +1,11 @@
1
-
+
+
src/main/resources/jobs/fixedLengthImportJob.xml
src/main/resources/jobs/multilineJob.xml