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

Removed the list of exceptions in favour of tx attributes (use +ExceptionType in config file)

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

@@ -26,6 +26,7 @@ import org.springframework.batch.item.validator.Validator;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.interceptor.TransactionAttribute;
 import org.springframework.util.Assert;
 
@@ -196,6 +197,14 @@ public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAw
 		this.transactionAttribute = transactionAttribute;
 	}
 
+	/**
+	 * Protected getter for the {@link TransactionAttribute} for subclasses only.
+	 * @return the transactionAttribute
+	 */
+	protected TransactionAttribute getTransactionAttribute() {
+		return transactionAttribute!=null?transactionAttribute:new DefaultTransactionAttribute();
+	}
+
 	/**
 	 * Create a {@link Step} from the configuration provided.
 	 * 

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

@@ -101,7 +101,7 @@ public class ItemOrientedStep extends AbstractStep {
 	public void setTransactionManager(PlatformTransactionManager transactionManager) {
 		this.transactionManager = transactionManager;
 	}
-	
+
 	/**
 	 * Public setter for the {@link TransactionAttribute}.
 	 * @param transactionAttribute the {@link TransactionAttribute} to set
@@ -109,7 +109,7 @@ public class ItemOrientedStep extends AbstractStep {
 	public void setTransactionAttribute(TransactionAttribute transactionAttribute) {
 		this.transactionAttribute = transactionAttribute;
 	}
-	
+
 	/**
 	 * Public setter for the {@link ItemHandler}.
 	 * 
@@ -245,7 +245,18 @@ public class ItemOrientedStep extends AbstractStep {
 
 				try {
 
-					exitStatus = processChunk(stepExecution, contribution);
+					try {
+						exitStatus = processChunk(stepExecution, contribution);
+					} catch (Error e) {
+						if (transactionAttribute.rollbackOn(e)) {
+							throw e;
+						}
+					} catch (Exception e) {
+						if (transactionAttribute.rollbackOn(e)) {
+							throw e;
+						}						
+					}
+
 					contribution.incrementCommitCount();
 
 					// If the step operations are asynchronous then we need
@@ -265,9 +276,30 @@ public class ItemOrientedStep extends AbstractStep {
 
 					// Attempt to flush before the step execution and stream
 					// state are updated
-					itemHandler.flush();
+					try {
+						itemHandler.flush();
+					} catch (Error e) {
+						if (transactionAttribute.rollbackOn(e)) {
+							throw e;
+						}
+					} catch (Exception e) {
+						if (transactionAttribute.rollbackOn(e)) {
+							throw e;
+						}						
+					}
+
+					try {
+						stream.update(stepExecution.getExecutionContext());
+					} catch (Error e) {
+						if (transactionAttribute.rollbackOn(e)) {
+							throw e;
+						}
+					} catch (Exception e) {
+						if (transactionAttribute.rollbackOn(e)) {
+							throw e;
+						}						
+					}
 
-					stream.update(stepExecution.getExecutionContext());
 					try {
 						getJobRepository().saveOrUpdateExecutionContext(stepExecution);
 					}

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

@@ -41,7 +41,7 @@ import org.springframework.batch.support.SubclassExceptionClassifier;
  * 
  * Skippable exceptions on write will by default cause transaction rollback - to
  * avoid rollback for specific exception class include it in the
- * {@link #setNoRollbackForExceptionClasses(Class[])} list.
+ * transaction attribute as "no rollback for".
  * 
  * @see SimpleStepFactoryBean
  * 
@@ -57,8 +57,6 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 
 	private Class[] fatalExceptionClasses = new Class[] { Error.class };
 
-	private Class[] noRollbackForExceptionClasses = new Class[] {};
-
 	private ItemKeyGenerator itemKeyGenerator;
 
 	private int cacheCapacity = 0;
@@ -169,18 +167,6 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 		this.itemKeyGenerator = itemKeyGenerator;
 	}
 
-	/**
-	 * Skippable noRollbackForExceptionClasses will *not* cause transaction
-	 * rollback.
-	 * 
-	 * @param noRollbackForExceptionClasses empty by default
-	 * 
-	 * @see #setSkippableExceptionClasses(Class[])
-	 */
-	public void setNoRollbackForExceptionClasses(Class[] noRollbackForExceptionClasses) {
-		this.noRollbackForExceptionClasses = noRollbackForExceptionClasses;
-	}
-
 	/**
 	 * Uses the {@link #setSkipLimit(int)} value to configure item handler and
 	 * and exception handler.
@@ -218,8 +204,11 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			getStepOperations().setExceptionHandler(
 					new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), fatalExceptionClasses));
 
-			RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy);
-			recoveryCallbackRetryPolicy.setRecoverableExceptionClasses(noRollbackForExceptionClasses);
+			RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy) {
+				protected boolean recoverForException(Throwable ex) {
+					return !getTransactionAttribute().rollbackOn(ex);
+				}
+			};
 			if (cacheCapacity > 0) {
 				recoveryCallbackRetryPolicy.setRetryContextCache(new MapRetryContextCache(cacheCapacity));
 			}