OPEN - BATCH-554: *StepFactoryBeans should have a list of exceptions that do and don't cause rollback.

added txValidExceptionClasses property to SkipLimitStepFactoryBean (passed to ItemSkipPolicyItemHandler#doNotRethrowExceptionClasses)

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;
+	}
+
 }

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);

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.
 	 */

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);