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 664dba8c6..1e7833a49 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 @@ -38,7 +38,13 @@ import org.springframework.transaction.support.TransactionSynchronizationManager /** * {@link ItemHandler} that implements skip behavior. It delegates to - * {@link #setItemSkipPolicy(ItemSkipPolicy)} to decide whether skip should be called or not. + * {@link #setItemSkipPolicy(ItemSkipPolicy)} to decide whether skip should be + * called or not. + * + * When exception is skipped on read it is *not* re-thrown (does not cause tx + * rollback). Skipped exception on write is re-thrown by default (causes tx + * rollback) unless the exception class is included in + * {@link #setDoNotRethrowExceptionClasses(Class[])}. * * If exception is thrown while reading the item, skip is called on the * {@link ItemReader}. If exception is thrown while writing the item, skip is @@ -63,6 +69,8 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { private Map skippedExceptions = new HashMap(); + private Class[] doNotRethrowExceptionClasses = new Class[] {}; + private ItemKeyGenerator defaultItemKeyGenerator = new ItemKeyGenerator() { public Object getKey(Object item) { return item; @@ -230,12 +238,27 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { // roll back addSkippedException(key, e); logger.debug("Added item to skip list; key=" + key); + + // return without re-throwing if exception shouldn't cause + // rollback + if (!shouldRethrow(e)) { + return; + } } - // always re-throw exception on write + // re-throw exception on write by default throw e; } } + private boolean shouldRethrow(Exception e) { + for (int i = 0; i < doNotRethrowExceptionClasses.length; i++) { + if (doNotRethrowExceptionClasses[i].isAssignableFrom(e.getClass())) { + return false; + } + } + return true; + } + public void mark() throws MarkFailedException { super.mark(); clearSkippedExceptions(); @@ -298,4 +321,12 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { } } + /** + * doNotRethrowExceptionClasses will not be re-thrown when skipped. + * @param doNotRethrowExceptionClasses empty by default + */ + public void setDoNotRethrowExceptionClasses(Class[] doNotRethrowExceptionClasses) { + this.doNotRethrowExceptionClasses = doNotRethrowExceptionClasses; + } + } 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 b7f41cc8d..786ed2fb8 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 @@ -33,6 +33,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { private Class[] skippableExceptionClasses = new Class[] { Exception.class }; private Class[] fatalExceptionClasses = new Class[] { Error.class }; + + private Class[] txValidExceptionClasses = new Class[] {}; private ItemKeyGenerator itemKeyGenerator; @@ -125,6 +127,17 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { public void setSkipCacheCapacity(int skipCacheCapacity) { this.skipCacheCapacity = skipCacheCapacity; } + + /** + * Skippable txValidExceptionClasses will *not* cause transaction rollback. + * + * @param txValidExceptionClasses empty by default + * + * @see #setSkippableExceptionClasses(Class[]) + */ + public void setTxValidExceptionClasses(Class[] txValidExceptionClasses) { + this.txValidExceptionClasses = txValidExceptionClasses; + } /** * Uses the {@link #setSkipLimit(int)} value to configure item handler and and @@ -148,6 +161,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { LimitCheckingItemSkipPolicy limitCheckingSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, Arrays .asList(skippableExceptionClasses), fatalExceptionList); itemHandler.setItemSkipPolicy(limitCheckingSkipPolicy); + itemHandler.setDoNotRethrowExceptionClasses(txValidExceptionClasses); this.itemSkipPolicy = limitCheckingSkipPolicy; SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler(skipLimit); exceptionHandler.setExceptionClasses(skippableExceptionClasses); 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 aed881876..310d832cf 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 @@ -124,11 +124,12 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { handler.handle(contribution); contribution.combineSkipCounts(); assertEquals(2, contribution.getSkipCount()); - // 2 is skipped so 3 was last one processed and now we are at 4, which was previously skipped + // 2 is skipped so 3 was last one processed and now we are at 4, which + // was previously skipped handler.handle(contribution); assertEquals(null, handler.read(contribution)); assertEquals(2, contribution.getSkipCount()); - + assertEquals(1, TransactionSynchronizationManager.getResourceMap().size()); Set removed = (Set) TransactionSynchronizationManager.getResourceMap().values().iterator().next(); // one skipped item was detected on read @@ -199,7 +200,7 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { catch (UnexpectedJobExecutionException e) { // expected String message = e.getMessage(); - assertTrue("Message does not contain 'capacity': "+message, message.indexOf("capacity")>=0); + 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 @@ -207,6 +208,30 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { assertEquals(new Holder("5"), handler.read(contribution)); } + /** + * Skippable write exceptions are not re-thrown when included in the + * {@link ItemSkipPolicyItemHandler#setDoNotRethrowExceptionClasses(Class[])} + */ + public void testWriteWithSkipAndDoNotRethrow() throws Exception { + + handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + handler.setDoNotRethrowExceptionClasses(new Class[] { SkippableException.class }); + + handler.handle(contribution); + handler.handle(contribution); + contribution.combineSkipCounts(); + assertEquals(1, contribution.getSkipCount()); + + // skippable exception thrown in writer at this point, but it won't be + // re-thrown + handler.handle(contribution); + + 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-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java index fe7079b4e..19722af5b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java @@ -34,19 +34,19 @@ import org.springframework.util.StringUtils; */ public class SkipLimitStepFactoryBeanTests extends TestCase { - SkipLimitStepFactoryBean factory = new SkipLimitStepFactoryBean(); + private SkipLimitStepFactoryBean factory = new SkipLimitStepFactoryBean(); - Class[] skippableExceptions = new Class[] { SkippableException.class, SkippableRuntimeException.class }; + private Class[] skippableExceptions = new Class[] { SkippableException.class, SkippableRuntimeException.class }; - final int SKIP_LIMIT = 2; + private final int SKIP_LIMIT = 2; - final int COMMIT_INTERVAL = 2; + private final int COMMIT_INTERVAL = 2; - SkipReaderStub reader = new SkipReaderStub(); + private SkipReaderStub reader = new SkipReaderStub(); - SkipWriterStub writer = new SkipWriterStub(); + private SkipWriterStub writer = new SkipWriterStub(); - JobExecution jobExecution; + private JobExecution jobExecution; protected int count; @@ -75,6 +75,9 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { assertEquals(2, stepExecution.getSkipCount()); + // only write exception caused rollback + assertEquals(1, stepExecution.getRollbackCount().intValue()); + // writer did not skip "2" as it never made it to writer, only "4" did assertTrue(reader.processed.contains("4")); assertFalse(writer.written.contains("4")); @@ -84,6 +87,24 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { } + /** + * Check skippable write exception does not cause rollback when included on + * {@link SkipLimitStepFactoryBean#setTxValidExceptionClasses(Class[])}. + */ + public void testSkipWithoutRethrow() throws Exception { + factory.setTxValidExceptionClasses(new Class[] { SkippableRuntimeException.class }); + AbstractStep step = (AbstractStep) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + step.execute(stepExecution); + + assertEquals(2, stepExecution.getSkipCount()); + + // no rollbacks + assertEquals(0, stepExecution.getRollbackCount().intValue()); + + } + /** * Fatal exception should cause immediate termination regardless of other * skip settings (note the fatal exception is also classified as skippable). @@ -227,7 +248,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { } public void testDefaultSkipPolicy() throws Exception { - factory.setSkippableExceptionClasses(new Class[] {Exception.class}); + factory.setSkippableExceptionClasses(new Class[] { Exception.class }); factory.setSkipLimit(1); List items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(new String[] { "a", "b", "c" })); @@ -246,7 +267,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase { StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); step.execute(stepExecution); - + assertEquals(1, stepExecution.getSkipCount()); // b is processed once and skipped, plus 1, plus c, plus the null at end assertEquals(4, count);