RESOLVED - issue BATCH-486: Users must choose between skip and retry

Tweak the retry policy to account for fatal exceptions as well (they should not be retryable).

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

@@ -68,6 +68,14 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 		this.fatalExceptionClasses = fatalExceptionClasses;
 	}
 
+	/**
+	 * Protected getter for the fatal exceptions.
+	 * @return the fatalExceptionClasses
+	 */
+	protected Class[] getFatalExceptionClasses() {
+		return fatalExceptionClasses;
+	}
+
 	/**
 	 * Public setter for the {@link ItemKeyGenerator}. This is used to identify
 	 * failed items so they can be skipped if encountered again, generally in

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

@@ -119,6 +119,7 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean {
 			SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(retryLimit);
 			if (retryableExceptionClasses != null) {
 				retryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
+				retryPolicy.setFatalExceptionClasses(getFatalExceptionClasses());
 			}
 
 			// Co-ordinate the retry policy with the exception handler:

spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java

@@ -48,7 +48,9 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 
 	private volatile int maxAttempts;
 
-	private BinaryExceptionClassifier classifier = new BinaryExceptionClassifier();
+	private BinaryExceptionClassifier retryableClassifier = new BinaryExceptionClassifier();
+
+	private BinaryExceptionClassifier fatalClassifier = new BinaryExceptionClassifier();
 
 	/**
 	 * Create a {@link SimpleRetryPolicy} with the default number of retry
@@ -67,6 +69,7 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 	public SimpleRetryPolicy(int maxAttempts) {
 		super();
 		setRetryableExceptionClasses(new Class[] { Exception.class });
+		setFatalExceptionClasses(new Class[] { Error.class });
 		this.maxAttempts = maxAttempts;
 	}
 
@@ -97,12 +100,23 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 
 	/**
 	 * Set the retryable exceptions. Any exception on the list, or subclasses
-	 * thereof, will be retryable. Others will be rethrown without retry.
+	 * thereof, will be retryable. Others will be re-thrown without retry.
 	 * 
 	 * @param retryableExceptionClasses defaults to {@link Exception}.
 	 */
 	public final void setRetryableExceptionClasses(Class[] retryableExceptionClasses) {
-		classifier.setExceptionClasses(retryableExceptionClasses);
+		retryableClassifier.setExceptionClasses(retryableExceptionClasses);
+	}
+
+	/**
+	 * Set the fatal exceptions. Any exception on the list, or subclasses
+	 * thereof, will be re-thrown without retry. This list takes precedence over
+	 * the retryable list.
+	 * 
+	 * @param retryableExceptionClasses defaults to {@link Exception}.
+	 */
+	public final void setFatalExceptionClasses(Class[] fatalExceptionClasses) {
+		fatalClassifier.setExceptionClasses(fatalExceptionClasses);
 	}
 
 	/**
@@ -126,7 +140,8 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 	 * Get a status object that can be used to track the current operation
 	 * according to this policy. Has to be aware of the latest exception and the
 	 * number of attempts.
-	 * @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback, RetryContext)
+	 * @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback,
+	 * RetryContext)
 	 */
 	public RetryContext open(RetryCallback callback, RetryContext parent) {
 		return new SimpleRetryContext(parent);
@@ -146,6 +161,6 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 	 * retryable.
 	 */
 	private boolean retryForException(Throwable ex) {
-		return !classifier.isDefault(ex);
+		return fatalClassifier.isDefault(ex) && !retryableClassifier.isDefault(ex);
 	}
 }

spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java

@@ -82,6 +82,24 @@ public class SimpleRetryPolicyTests extends TestCase {
 		assertEquals("foo", context.getLastThrowable().getMessage());
 	}
 
+	public void testDefaultFatal() throws Exception {
+		SimpleRetryPolicy policy = new SimpleRetryPolicy();
+		RetryContext context = policy.open(null, null);
+		assertNotNull(context);
+		policy.registerThrowable(context, new Error("foo"));
+		assertFalse(policy.canRetry(context));
+	}
+
+	public void testFatalOverridesRetryable() throws Exception {
+		SimpleRetryPolicy policy = new SimpleRetryPolicy();
+		policy.setFatalExceptionClasses(new Class[] {Exception.class});
+		policy.setRetryableExceptionClasses(new Class[] {RuntimeException.class});
+		RetryContext context = policy.open(null, null);
+		assertNotNull(context);
+		policy.registerThrowable(context, new RuntimeException("foo"));
+		assertFalse(policy.canRetry(context));
+	}
+
 	public void testParent() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null, null);