RESOLVED - issue BATCH-485: SkippableExceptionClasses should be inverted

http://jira.springframework.org/browse/BATCH-485

SkipLimitStepFactoryBean now declares setter for fatal exceptions that are never skipped and immediately rethrown by exception handlerRESOLVED - issue BATCH-485: SkippableExceptionClasses should be inverted 
http://jira.springframework.org/browse/BATCH-485

SkipLimitStepFactoryBean now declares setter for fatal exceptions that are never skipped and immediately rethrown by exception handler.
LimitCheckingItemSkipPolicy and SimpleLimitExceptionHandler were modified to recognize fatal exceptions.

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

@@ -7,15 +7,28 @@ import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
 import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
 
 /**
- * Factory bean for step that allows configuring skip limit.
+ * Factory bean for step that provides options for configuring skip behavior.
+ * User can set {@link #skipLimit} to set how many exceptions of
+ * {@link #skippableExceptionClasses} types are tolerated.
+ * {@link #fatalExceptionClasses} will cause immediate termination of job - they
+ * are treated as higher priority than {@link #skippableExceptionClasses}, so
+ * the two lists don't need to be exclusive.
+ * 
+ * @see SimpleStepFactoryBean
+ * @see StatefulRetryStepFactoryBean
+ * 
+ * @author Dave Syer
+ * @author Robert Kasanicky
  * 
  */
 public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 
 	private int skipLimit = 0;
-	
-	private Class[] skippableExceptionClasses = new Class[]{ Exception.class };
-	
+
+	private Class[] skippableExceptionClasses = new Class[] { Exception.class };
+
+	private Class[] fatalExceptionClasses = new Class[] { Error.class };
+
 	/**
 	 * 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
@@ -28,11 +41,11 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 	public void setSkipLimit(int skipLimit) {
 		this.skipLimit = skipLimit;
 	}
-	
+
 	/**
 	 * Public setter for exception classes that when raised won't crash the job
-	 * but will result in transaction rollback and the item which handling caused
-	 * the exception will be skipped.
+	 * but will result in transaction rollback and the item which handling
+	 * caused the exception will be skipped.
 	 * 
 	 * @param skippableExceptionClasses defaults to <code>Exception</code>
 	 */
@@ -41,12 +54,21 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 	}
 
 	/**
-	 * Uses the {@link #skipLimit} value to configure item handler and
-	 * and exception handler.
+	 * Public setter for exception classes that should cause immediate failure.
+	 * 
+	 * @param fatalExceptionClasses {@link Error} by default
+	 */
+	public void setFatalExceptionClasses(Class[] fatalExceptionClasses) {
+		this.fatalExceptionClasses = fatalExceptionClasses;
+	}
+
+	/**
+	 * Uses the {@link #skipLimit} value to configure item handler and and
+	 * exception handler.
 	 */
 	protected void applyConfiguration(ItemOrientedStep step) {
 		super.applyConfiguration(step);
-		
+
 		ItemSkipPolicyItemHandler itemHandler = new ItemSkipPolicyItemHandler(getItemReader(), getItemWriter());
 
 		if (skipLimit > 0) {
@@ -55,10 +77,12 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			 * to absorb exceptions at the step level because the failed items
 			 * will never re-appear after a rollback.
 			 */
-			itemHandler.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(skipLimit, Arrays.asList(skippableExceptionClasses)));
+			itemHandler.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(skipLimit, Arrays
+					.asList(skippableExceptionClasses), Arrays.asList(fatalExceptionClasses)));
 			SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler();
 			exceptionHandler.setLimit(skipLimit);
 			exceptionHandler.setExceptionClasses(skippableExceptionClasses);
+			exceptionHandler.setFatalExceptionClasses(fatalExceptionClasses);
 			setExceptionHandler(exceptionHandler);
 			getStepOperations().setExceptionHandler(getExceptionHandler());
 		}
@@ -66,8 +90,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			// This is the default in ItemOrientedStep anyway...
 			itemHandler.setItemSkipPolicy(new NeverSkipItemSkipPolicy());
 		}
-		
+
 		step.setItemHandler(itemHandler);
 	}
-	
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java

@@ -45,12 +45,16 @@ import org.springframework.batch.support.SubclassExceptionClassifier;
  * likely want to skip, but a {@link FileNotFoundException} should cause
  * immediate termination of the {@link Step}. Because it would be impossible
  * for a general purpose policy to determine all the types of exceptions that
- * should be skipped from those that shouldn't, a list must be passed in, with
- * all of the exceptions that are 'skippable'
+ * should be skipped from those that shouldn't, two lists must be passed in,
+ * with all of the exceptions that are 'fatal' and 'skippable'. The two lists
+ * are not enforced to be exclusive, they are prioritized instead - exceptions
+ * that are fatal will never be skipped, regardless whether the exception can
+ * also be classified as skippable.
  * </p>
  * 
  * @author Ben Hale
  * @author Lucas Ward
+ * @author Robert Kasanicky
  */
 public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 
@@ -59,15 +63,33 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 	 */
 	private static final String SKIP = "skip";
 
+	/**
+	 * Label for classifying fatal exceptions - these are never skipped.
+	 */
+	private static final String NEVER_SKIP = "neverSkip";
+
 	private final int skipLimit;
 
 	private ExceptionClassifier exceptionClassifier;
 
+	/**
+	 * Convenience constructor that assumes all exception types are skippable
+	 * and none are fatal.
+	 * @param skipLimit the number of exceptions allowed to skip
+	 */
 	public LimitCheckingItemSkipPolicy(int skipLimit) {
-		this(skipLimit, Collections.singletonList(Exception.class));
+		this(skipLimit, Collections.singletonList(Exception.class), Collections.EMPTY_LIST);
 	}
 
-	public LimitCheckingItemSkipPolicy(int skipLimit, List skippableExceptions) {
+	/**
+	 * 
+	 * @param skipLimit the number of skippable exceptions that are allowed to
+	 * be skipped
+	 * @param skippableExceptions exception classes that can be skipped
+	 * (non-critical)
+	 * @param fatalExceptions exception classes that should never be skipped
+	 */
+	public LimitCheckingItemSkipPolicy(int skipLimit, List skippableExceptions, List fatalExceptions) {
 		this.skipLimit = skipLimit;
 		SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
 		Map typeMap = new HashMap();
@@ -75,6 +97,10 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 			Class throwable = (Class) iterator.next();
 			typeMap.put(throwable, SKIP);
 		}
+		for (Iterator iterator = fatalExceptions.iterator(); iterator.hasNext();) {
+			Class throwable = (Class) iterator.next();
+			typeMap.put(throwable, NEVER_SKIP);
+		}
 		exceptionClassifier.setTypeMap(typeMap);
 		this.exceptionClassifier = exceptionClassifier;
 	}
@@ -82,12 +108,16 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 	/**
 	 * Given the provided exception and skip count, determine whether or not
 	 * processing should continue for the given exception. If the exception is
-	 * not within the list of 'skippable exceptions', false will be returned. If
-	 * the exception is within the list, and {@link StepExecution} skipCount is
-	 * greater than the skipLimit, then a {@link SkipLimitExceededException}
-	 * will be thrown.
+	 * not within the list of 'skippable exceptions' or belongs to the list of
+	 * 'fatal exceptions', false will be returned. If the exception is within
+	 * the skippable list (and not in the fatal list), and {@link StepExecution}
+	 * skipCount is greater than the skipLimit, then a
+	 * {@link SkipLimitExceededException} will be thrown.
 	 */
 	public boolean shouldSkip(Throwable t, int skipCount) {
+		if (exceptionClassifier.classify(t).equals(NEVER_SKIP)) {
+			return false;
+		}
 		if (exceptionClassifier.classify(t).equals(SKIP)) {
 			if (skipCount < skipLimit) {
 				return true;