From 4ce2a39f303cd67734f723a3f4b76899dd5000b2 Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 3 Apr 2008 11:09:13 +0000 Subject: [PATCH] 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. --- .../step/item/ItemSkipPolicyItemHandler.java | 24 ++++++++++++++++++ .../step/item/SkipLimitStepFactoryBean.java | 25 +++++++++++++++++-- .../item/ItemSkipPolicyItemHandlerTests.java | 24 ++++++++++++++++++ spring-batch-samples/.springBeans | 3 ++- 4 files changed, 73 insertions(+), 3 deletions(-) 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