BATCH-220: Rationalise the exception classifiers so they can be used in a retry

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java

@@ -52,7 +52,7 @@ public class ExecutionContext implements Serializable {
 	 * @param map Initial contents of context.
 	 */
 	public ExecutionContext(Map<String, Object> map) {
-		this.map = map;
+		this.map = new HashMap<String, Object>(map);
 	}
 
 	/**

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/context/RepeatContextCounter.java

@@ -33,15 +33,15 @@ import org.springframework.util.Assert;
  */
 public class RepeatContextCounter {
 
-	private String countKey;
+	final private String countKey;
 
 	/**
 	 * Flag to indicate whether the count is stored at the level of the parent
 	 * context, or just local to the current context. Default value is false.
 	 */
-	private boolean useParent = false;
+	final private boolean useParent;
 
-	private RepeatContext context;
+	final private RepeatContext context;
 
 	/**
 	 * Increment the counter.
@@ -82,7 +82,7 @@ public class RepeatContextCounter {
 
 		super();
 		
-		Assert.notNull(context, "The context must be provided");
+		Assert.notNull(context, "The context must be provided to initialize a counter");
 
 		this.countKey = countKey;
 		this.useParent = useParent;

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/LogOrRethrowExceptionHandler.java

@@ -21,12 +21,12 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.RepeatException;
 import org.springframework.batch.support.Classifier;
-import org.springframework.batch.support.ExceptionClassifierSupport;
+import org.springframework.batch.support.ClassifierSupport;
 
 /**
- * Implementation of {@link ExceptionHandler} based on an {@link Classifier}. The classifier determines
- * whether to log the exception or rethrow it. The keys in the classifier must be the same as the static contants in
- * this class.
+ * Implementation of {@link ExceptionHandler} based on an {@link Classifier}.
+ * The classifier determines whether to log the exception or rethrow it. The
+ * keys in the classifier must be the same as the static enum in this class.
  * 
  * @author Dave Syer
  * 
@@ -34,46 +34,57 @@ import org.springframework.batch.support.ExceptionClassifierSupport;
 public class LogOrRethrowExceptionHandler implements ExceptionHandler {
 
 	/**
-	 * Key for {@link Classifier} signalling that the throwable should be rethrown. If the throwable is not a
-	 * RuntimeException it is wrapped in a {@link RepeatException}.
+	 * Logging levels for the handler.
+	 * 
+	 * @author Dave Syer
+	 * 
 	 */
-	public static final String RETHROW = "rethrow";
+	public static enum Level {
 
-	/**
-	 * Key for {@link Classifier} signalling that the throwable should be logged at debug level.
-	 */
-	public static final String DEBUG = "debug";
+		/**
+		 * Key for {@link Classifier} signalling that the throwable should be
+		 * rethrown. If the throwable is not a RuntimeException it is wrapped in
+		 * a {@link RepeatException}.
+		 */
+		RETHROW,
 
-	/**
-	 * Key for {@link Classifier} signalling that the throwable should be logged at warn level.
-	 */
-	public static final String WARN = "warn";
+		/**
+		 * Key for {@link Classifier} signalling that the throwable should be
+		 * logged at debug level.
+		 */
+		DEBUG,
 
-	/**
-	 * Key for {@link Classifier} signalling that the throwable should be logged at error level.
-	 */
-	public static final String ERROR = "error";
+		/**
+		 * Key for {@link Classifier} signalling that the throwable should be
+		 * logged at warn level.
+		 */
+		WARN,
+
+		/**
+		 * Key for {@link Classifier} signalling that the throwable should be
+		 * logged at error level.
+		 */
+		ERROR
+
+	}
 
 	protected final Log logger = LogFactory.getLog(LogOrRethrowExceptionHandler.class);
 
-	private Classifier<Throwable, String> exceptionClassifier = new ExceptionClassifierSupport() {
-		public String classify(Throwable throwable) {
-			return RETHROW;
-		}
-	};
+	private Classifier<Throwable, Level> exceptionClassifier = new ClassifierSupport<Throwable, Level>(Level.RETHROW);
 
 	/**
-	 * Setter for the {@link Classifier} used by this handler. The default is to map all throwable instances to
-	 * {@link #RETHROW}.
+	 * Setter for the {@link Classifier} used by this handler. The default is to
+	 * map all throwable instances to {@link Level#RETHROW}.
 	 * 
 	 * @param exceptionClassifier the ExceptionClassifier to use
 	 */
-	public void setExceptionClassifier(Classifier<Throwable,String> exceptionClassifier) {
+	public void setExceptionClassifier(Classifier<Throwable, Level> exceptionClassifier) {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 
 	/**
-	 * Classify the throwables and decide whether to rethrow based on the result. The context is not used.
+	 * Classify the throwables and decide whether to rethrow based on the
+	 * result. The context is not used.
 	 * 
 	 * @throws Throwable
 	 * 
@@ -81,18 +92,18 @@ public class LogOrRethrowExceptionHandler implements ExceptionHandler {
 	 */
 	public void handleException(RepeatContext context, Throwable throwable) throws Throwable {
 
-		String key = exceptionClassifier.classify(throwable);
-		if (ERROR.equals(key)) {
+		Level key = exceptionClassifier.classify(throwable);
+		if (Level.ERROR.equals(key)) {
 			logger.error("Exception encountered in batch repeat.", throwable);
-		} else if (WARN.equals(key)) {
+		}
+		else if (Level.WARN.equals(key)) {
 			logger.warn("Exception encountered in batch repeat.", throwable);
-		} else if (DEBUG.equals(key) && logger.isDebugEnabled()) {
+		}
+		else if (Level.DEBUG.equals(key) && logger.isDebugEnabled()) {
 			logger.debug("Exception encountered in batch repeat.", throwable);
-		} else if (RETHROW.equals(key)) {
+		}
+		else if (Level.RETHROW.equals(key)) {
 			throw throwable;
-		} else {
-			throw new IllegalStateException(
-			        "Unclassified exception encountered.  Did you mean to classifiy this as 'rethrow'?");
 		}
 
 	}

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandler.java

@@ -24,24 +24,27 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.context.RepeatContextCounter;
 import org.springframework.batch.support.Classifier;
-import org.springframework.batch.support.ExceptionClassifierSupport;
+import org.springframework.batch.support.SubclassClassifier;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Implementation of {@link ExceptionHandler} that rethrows when exceptions of a
- * given type reach a threshold. Requires an {@link Classifier} that
- * maps exception types to unique keys, and also a map from those keys to
- * threshold values (Integer type).
+ * given type reach a threshold. Requires an {@link Classifier} that maps
+ * exception types to unique keys, and also a map from those keys to threshold
+ * values (Integer type).
  * 
  * @author Dave Syer
  * 
  */
 public class RethrowOnThresholdExceptionHandler implements ExceptionHandler {
 
+	protected static final IntegerHolder ZERO = new IntegerHolder(0);
+
 	protected final Log logger = LogFactory.getLog(RethrowOnThresholdExceptionHandler.class);
 
-	private Classifier<Throwable,String> exceptionClassifier = new ExceptionClassifierSupport();
-
-	private Map<String, Integer> thresholds = new HashMap<String, Integer>();
+	private Classifier<? super Throwable, IntegerHolder> exceptionClassifier = new Classifier<Throwable, IntegerHolder>() {
+		public RethrowOnThresholdExceptionHandler.IntegerHolder classify(Throwable classifiable) { return ZERO;}
+	};
 
 	private boolean useParent = false;
 
@@ -63,30 +66,19 @@ public class RethrowOnThresholdExceptionHandler implements ExceptionHandler {
 	 */
 	public RethrowOnThresholdExceptionHandler() {
 		super();
-		thresholds.put(ExceptionClassifierSupport.DEFAULT, 0);
 	}
 
 	/**
-	 * A map from classifier keys to a threshold value of type Integer. The keys
-	 * are usually String literals, depending on the {@link Classifier}
-	 * implementation used.
+	 * A map from exception classes to a threshold value of type Integer.
 	 * 
 	 * @param thresholds the threshold value map.
 	 */
-	public void setThresholds(Map<String, Integer> thresholds) {
-		this.thresholds = thresholds;
-	}
-
-	/**
-	 * Setter for the {@link Classifier} used by this handler. The
-	 * default is to map all throwable instances to
-	 * {@link ExceptionClassifierSupport#DEFAULT}, which are then mapped to a
-	 * threshold of 0 by the {@link #setThresholds(Map)} map.
-	 * 
-	 * @param exceptionClassifier ExceptionClassifier to use
-	 */
-	public void setExceptionClassifier(Classifier<Throwable, String> exceptionClassifier) {
-		this.exceptionClassifier = exceptionClassifier;
+	public void setThresholds(Map<Class<? extends Throwable>, Integer> thresholds) {
+		Map<Class<? extends Throwable>, IntegerHolder> typeMap = new HashMap<Class<? extends Throwable>, IntegerHolder>();
+		for (Class<? extends Throwable> type : thresholds.keySet()) {
+			typeMap.put(type, new IntegerHolder(thresholds.get(type)));
+		}
+		exceptionClassifier = new SubclassClassifier<Throwable, IntegerHolder>(typeMap, ZERO);
 	}
 
 	/**
@@ -99,21 +91,55 @@ public class RethrowOnThresholdExceptionHandler implements ExceptionHandler {
 	 */
 	public void handleException(RepeatContext context, Throwable throwable) throws Throwable {
 
-		Object key = exceptionClassifier.classify(throwable);
+		IntegerHolder key = exceptionClassifier.classify(throwable);
+		
 		RepeatContextCounter counter = getCounter(context, key);
 		counter.increment();
 		int count = counter.getCount();
-		Integer threshold = thresholds.get(key);
-		if (threshold == null || count > threshold) {
+		int threshold = key.getValue();
+		if (count > threshold) {
 			throw throwable;
 		}
 
 	}
 
-	private RepeatContextCounter getCounter(RepeatContext context, Object key) {
-		String attribute = RethrowOnThresholdExceptionHandler.class.getName() + "." + key.toString();
+	private RepeatContextCounter getCounter(RepeatContext context, IntegerHolder key) {
+		String attribute = RethrowOnThresholdExceptionHandler.class.getName() + "." + key;
 		// Creates a new counter and stores it in the correct context:
 		return new RepeatContextCounter(context, attribute, useParent);
 	}
 
+	/**
+	 * @author Dave Syer
+	 * 
+	 */
+	private static class IntegerHolder {
+
+		private final int value;
+
+		/**
+		 * @param value
+		 */
+		public IntegerHolder(int value) {
+			this.value = value;
+		}
+		
+		/**
+		 * Public getter for the value.
+		 * @return the value
+		 */
+		public int getValue() {
+			return value;
+		}
+		
+		/* (non-Javadoc)
+		 * @see java.lang.Object#toString()
+		 */
+		@Override
+		public String toString() {
+			return ObjectUtils.getIdentityHexString(this)+"."+value;
+		}
+
+	}
+
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/SimpleLimitExceptionHandler.java

@@ -16,10 +16,13 @@
 
 package org.springframework.batch.repeat.exception;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.Map;
 
 import org.springframework.batch.repeat.RepeatContext;
-import org.springframework.batch.support.ExceptionClassifierSupport;
+import org.springframework.beans.factory.InitializingBean;
 
 /**
  * Simple implementation of exception handler which looks for given exception
@@ -32,24 +35,37 @@ import org.springframework.batch.support.ExceptionClassifierSupport;
  * @author Dave Syer
  * @author Robert Kasanicky
  */
-public class SimpleLimitExceptionHandler implements ExceptionHandler {
-
-	/**
-	 * Name of exception classifier key for the nominated exception types.
-	 */
-	private static final String TX_INVALID = "TX_INVALID";
-
-	/**
-	 * Name of exception classifier key for the fatal exception types (not
-	 * counted, immediately rethrown).
-	 */
-	private static final String FATAL = "FATAL";
+public class SimpleLimitExceptionHandler implements ExceptionHandler, InitializingBean {
 
 	private RethrowOnThresholdExceptionHandler delegate = new RethrowOnThresholdExceptionHandler();
 
-	private Class<?>[] exceptionClasses = new Class[] { Exception.class };
+	private Collection<Class<? extends Throwable>> exceptionClasses = Collections
+			.<Class<? extends Throwable>> singleton(Exception.class);
 
-	private Class<?>[] fatalExceptionClasses = new Class[] { Error.class };
+	private Collection<Class<? extends Throwable>> fatalExceptionClasses = Collections
+			.<Class<? extends Throwable>> singleton(Error.class);
+
+	private int limit = 0;
+
+	/**
+	 * Apply the provided properties to create a delegate handler.
+	 * 
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		if (limit <= 0) {
+			return;
+		}
+		Map<Class<? extends Throwable>, Integer> thresholds = new HashMap<Class<? extends Throwable>, Integer>();
+		for (Class<? extends Throwable> type : exceptionClasses) {
+			thresholds.put(type, limit);
+		}
+		// do the fatalExceptionClasses last so they override the others
+		for (Class<? extends Throwable> type : fatalExceptionClasses) {
+			thresholds.put(type, 0);
+		}
+		delegate.setThresholds(thresholds);
+	}
 
 	/**
 	 * Flag to indicate the the exception counters should be shared between
@@ -67,12 +83,12 @@ public class SimpleLimitExceptionHandler implements ExceptionHandler {
 	/**
 	 * Convenience constructor for the {@link SimpleLimitExceptionHandler} to
 	 * set the limit.
-	 *
+	 * 
 	 * @param limit the limit
 	 */
 	public SimpleLimitExceptionHandler(int limit) {
 		this();
-		setLimit(limit);
+		this.limit = limit;
 	}
 
 	/**
@@ -80,28 +96,13 @@ public class SimpleLimitExceptionHandler implements ExceptionHandler {
 	 */
 	public SimpleLimitExceptionHandler() {
 		super();
-		delegate.setExceptionClassifier(new ExceptionClassifierSupport() {
-			public String classify(Throwable throwable) {
-				for (Class<?> fatalExceptionClass : fatalExceptionClasses) {
-					if (fatalExceptionClass.isAssignableFrom(throwable.getClass())) {
-						return FATAL;
-					}
-				}
-				for (Class<?> exceptionClass : exceptionClasses) {
-					if (exceptionClass.isAssignableFrom(throwable.getClass())) {
-						return TX_INVALID;
-					}
-				}
-				return super.classify(throwable);
-			}
-		});
 	}
 
 	/**
 	 * Rethrows only if the limit is breached for this context on the exception
 	 * type specified.
 	 * 
-	 * @see #setExceptionClasses(Class[])
+	 * @see #setExceptionClasses(Collection)
 	 * @see #setLimit(int)
 	 * 
 	 * @see org.springframework.batch.repeat.exception.ExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext,
@@ -118,34 +119,28 @@ public class SimpleLimitExceptionHandler implements ExceptionHandler {
 	 * @param limit the limit
 	 */
 	public void setLimit(final int limit) {
-		delegate.setThresholds(new HashMap<String, Integer>() {
-			{
-				put(ExceptionClassifierSupport.DEFAULT, 0);
-				put(TX_INVALID, limit);
-				put(FATAL, 0);
-			}
-		});
+		this.limit = limit;
 	}
 
 	/**
-	 * Setter for the Throwable exceptionClasses that this handler counts.
-	 * Defaults to {@link Exception}. If more exceptionClasses are specified
-	 * handler uses single counter that is incremented when one of the
-	 * recognized exception exceptionClasses is handled.
+	 * Setter for the exception classes that this handler counts. Defaults to
+	 * {@link Exception}. If more exceptionClasses are specified handler uses
+	 * single counter that is incremented when one of the recognized exception
+	 * exceptionClasses is handled.
 	 * @param classes exceptionClasses
 	 */
-	public void setExceptionClasses(Class<?>[] classes) {
+	public void setExceptionClasses(Collection<Class<? extends Throwable>> classes) {
 		this.exceptionClasses = classes;
 	}
 
 	/**
-	 * Setter for the Throwable exceptionClasses that shouldn't be counted, but
-	 * rethrown immediately. This list has higher priority than
-	 * {@link #setExceptionClasses(Class[])}.
+	 * Setter for the exception classes that shouldn't be counted, but rethrown
+	 * immediately. This list has higher priority than
+	 * {@link #setExceptionClasses(Collection)}.
 	 * 
 	 * @param fatalExceptionClasses defaults to {@link Error}
 	 */
-	public void setFatalExceptionClasses(Class<?>[] fatalExceptionClasses) {
+	public void setFatalExceptionClasses(Collection<Class<? extends Throwable>> fatalExceptionClasses) {
 		this.fatalExceptionClasses = fatalExceptionClasses;
 	}
 

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

@@ -23,7 +23,8 @@ import org.springframework.batch.retry.RetryContext;
 import org.springframework.batch.retry.RetryPolicy;
 import org.springframework.batch.retry.context.RetryContextSupport;
 import org.springframework.batch.support.Classifier;
-import org.springframework.batch.support.ExceptionClassifierSupport;
+import org.springframework.batch.support.ClassifierSupport;
+import org.springframework.batch.support.SubclassClassifier;
 import org.springframework.util.Assert;
 
 /**
@@ -35,34 +36,32 @@ import org.springframework.util.Assert;
  */
 public class ExceptionClassifierRetryPolicy implements RetryPolicy {
 
-	private Classifier<Throwable, String> exceptionClassifier = new ExceptionClassifierSupport();
-
-	private Map<String, RetryPolicy> policyMap = new HashMap<String, RetryPolicy>();
-
-	public ExceptionClassifierRetryPolicy() {
-		policyMap.put(ExceptionClassifierSupport.DEFAULT, new NeverRetryPolicy());
-	}
+	private Classifier<Throwable, RetryPolicy> exceptionClassifier = new ClassifierSupport<Throwable, RetryPolicy>(
+			new NeverRetryPolicy());
 
 	/**
 	 * Setter for policy map. This property should not be changed dynamically -
 	 * set it once, e.g. in configuration, and then don't change it during a
-	 * running application.
+	 * running application. Either this property or the exception classifier
+	 * directly should be set, but not both.
 	 * 
-	 * @param policyMap a map of String to {@link RetryPolicy} that will be
-	 * applied to the result of the {@link Classifier} to locate a
-	 * policy.
+	 * @param policyMap a map of String to {@link RetryPolicy} that will be used
+	 * to create a {@link Classifier} to locate a policy.
 	 */
-	public void setPolicyMap(Map<String, RetryPolicy> policyMap) {
-		this.policyMap = policyMap;
+	public void setPolicyMap(Map<Class<? extends Throwable>, RetryPolicy> policyMap) {
+		SubclassClassifier<Throwable, RetryPolicy> subclassClassifier = new SubclassClassifier<Throwable, RetryPolicy>(
+				policyMap, (RetryPolicy) new NeverRetryPolicy());
+		this.exceptionClassifier = subclassClassifier;
 	}
 
 	/**
 	 * Setter for an exception classifier. The classifier is responsible for
-	 * translating exceptions to keys in the policy map.
+	 * translating exceptions to concrete retry policies. Either this property
+	 * or the policy map should be used, but not both.
 	 * 
 	 * @param exceptionClassifier ExceptionClassifier to use
 	 */
-	public void setExceptionClassifier(Classifier<Throwable,String> exceptionClassifier) {
+	public void setExceptionClassifier(Classifier<Throwable, RetryPolicy> exceptionClassifier) {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 
@@ -110,22 +109,20 @@ public class ExceptionClassifierRetryPolicy implements RetryPolicy {
 
 	private class ExceptionClassifierRetryContext extends RetryContextSupport implements RetryPolicy {
 
-		private Classifier<Throwable, String> exceptionClassifier;
+		final private Classifier<Throwable, RetryPolicy> exceptionClassifier;
 
 		// Dynamic: depends on the latest exception:
-		RetryPolicy policy;
+		private RetryPolicy policy;
 
 		// Dynamic: depends on the policy:
-		RetryContext context;
+		private RetryContext context;
 
-		Map<RetryPolicy, RetryContext> contexts = new HashMap<RetryPolicy, RetryContext>();
+		final private Map<RetryPolicy, RetryContext> contexts = new HashMap<RetryPolicy, RetryContext>();
 
-		public ExceptionClassifierRetryContext(RetryContext parent, Classifier<Throwable,String> exceptionClassifier) {
+		public ExceptionClassifierRetryContext(RetryContext parent,
+				Classifier<Throwable, RetryPolicy> exceptionClassifier) {
 			super(parent);
 			this.exceptionClassifier = exceptionClassifier;
-			Object key = exceptionClassifier.getDefault();
-			policy = getPolicy(key);
-			Assert.notNull(policy, "Could not locate default policy: key=[" + key + "].");
 		}
 
 		public boolean canRetry(RetryContext context) {
@@ -139,7 +136,7 @@ public class ExceptionClassifierRetryPolicy implements RetryPolicy {
 		public void close(RetryContext context) {
 			// Only close those policies that have been used (opened):
 			for (RetryPolicy policy : contexts.keySet()) {
-				policy.close(getContext(policy));
+				policy.close(getContext(policy, context.getParent()));
 			}
 		}
 
@@ -148,26 +145,21 @@ public class ExceptionClassifierRetryPolicy implements RetryPolicy {
 		}
 
 		public void registerThrowable(RetryContext context, Exception throwable) {
-			policy = getPolicy(exceptionClassifier.classify(throwable));
-			this.context = getContext(policy);
+			policy = exceptionClassifier.classify(throwable);
+			Assert.notNull(policy, "Could not locate policy for exception=[" + throwable + "].");
+			this.context = getContext(policy, context.getParent());
 			policy.registerThrowable(this.context, throwable);
 		}
 
-		private RetryContext getContext(RetryPolicy policy) {
+		private RetryContext getContext(RetryPolicy policy, RetryContext parent) {
 			RetryContext context = contexts.get(policy);
 			if (context == null) {
-				context = policy.open(null);
+				context = policy.open(parent);
 				contexts.put(policy, context);
 			}
 			return context;
 		}
 
-		private RetryPolicy getPolicy(Object key) {
-			RetryPolicy result = policyMap.get(key);
-			Assert.notNull(result, "Could not locate policy for key=[" + key + "].");
-			return result;
-		}
-
 	}
 
 }

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

@@ -107,7 +107,7 @@ public class SimpleRetryPolicy implements RetryPolicy {
 	 * @param retryableExceptionClasses defaults to {@link Exception}.
 	 */
 	public final void setRetryableExceptionClasses(Collection<Class<? extends Throwable>> retryableExceptionClasses) {
-		retryableClassifier.setExceptionClasses(retryableExceptionClasses);
+		retryableClassifier.setTypes(retryableExceptionClasses);
 	}
 
 	/**
@@ -118,7 +118,7 @@ public class SimpleRetryPolicy implements RetryPolicy {
 	 * @param fatalExceptionClasses defaults to {@link Exception}.
 	 */
 	public final void setFatalExceptionClasses(Collection<Class<? extends Throwable>> fatalExceptionClasses) {
-		fatalClassifier.setExceptionClasses(fatalExceptionClasses);
+		fatalClassifier.setTypes(fatalExceptionClasses);
 	}
 
 	/**
@@ -162,6 +162,6 @@ public class SimpleRetryPolicy implements RetryPolicy {
 	 * retryable.
 	 */
 	private boolean retryForException(Throwable ex) {
-		return fatalClassifier.isDefault(ex) && !retryableClassifier.isDefault(ex);
+		return !fatalClassifier.classify(ex) && retryableClassifier.classify(ex);
 	}
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java

@@ -39,6 +39,7 @@ import org.springframework.batch.retry.backoff.NoBackOffPolicy;
 import org.springframework.batch.retry.policy.MapRetryContextCache;
 import org.springframework.batch.retry.policy.RetryContextCache;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
+import org.springframework.batch.support.Classifier;
 
 /**
  * Template class that simplifies the execution of operations with retry
@@ -78,6 +79,45 @@ public class RetryTemplate implements RetryOperations {
 
 	private RetryContextCache retryContextCache = new MapRetryContextCache();
 
+	private Classifier<? super Throwable, Boolean> rollbackClassifier = null;
+
+	/**
+	 * Public setter for the rollback classifier. This classifier answers its
+	 * default if the exception provided should not cause a rollback. I.e.
+	 * anything other than the default will lead to a rollback.<br/><br/>
+	 * 
+	 * The decision whether to rollback or not is unrelated to that of the
+	 * {@link RetryPolicy}, but the policy can be accidentally inconsistent
+	 * with the rollback decision. E.g. in a stateless retry the policy might be
+	 * configured to allow retry on a rollback, but that wouldn't make sense - a
+	 * stateful retry should have been used. The best we can do in such
+	 * circumstances is throw a {@link RetryException} from
+	 * {@link #execute(RetryCallback)} or
+	 * {@link #execute(RetryCallback, RecoveryCallback)}. The recovery path
+	 * will not be taken in such situations.<br/><br/>
+	 * 
+	 * For stateless retry it is often adequate to use the default behaviour, as
+	 * long as one is careful with the retry policy (exceptions which should
+	 * cause rollback are still not really retryable in a transactional
+	 * setting).<br/><br/>
+	 * 
+	 * For stateful retry adding a classifier will allow an optimisation:
+	 * exceptions which are not marked for rollback can still be retried, but
+	 * without paying the cost of a rollback. Effectively one is overriding the
+	 * stateful quality of the retry dynamically, according to the exception
+	 * type.<br/><br/>
+	 * 
+	 * Example usage would be for a stateful retry to specify a validation exception as not for rollback
+	 * 
+	 * If not set then the default is to rollback for all exceptions when the
+	 * retry is stateful, and for none when it is stateless.
+	 * 
+	 * @param rollbackClassifier the rollback classifier to set
+	 */
+	public void setRollbackClassifier(Classifier<? super Throwable, Boolean> rollbackClassifier) {
+		this.rollbackClassifier = rollbackClassifier;
+	}
+
 	/**
 	 * Public setter for the {@link RetryContextCache}.
 	 * @param retryContextCache the {@link RetryContextCache} to set.
@@ -406,7 +446,17 @@ public class RetryTemplate implements RetryOperations {
 	 * otherwise
 	 */
 	protected boolean shouldRethrow(RetryPolicy retryPolicy, RetryContext context, RetryState state) {
-		// TODO: allow stateless behaviour to take over for certain exception types
+		// Allow stateless behaviour to take over for certain exception types
+		if (rollbackClassifier != null) {
+			boolean rollback = rollbackClassifier.classify(context.getLastThrowable());
+			if (rollback && state == null && retryPolicy.canRetry(context)) {
+				throw new RetryException("Inconsistent configuration.  The retry policy says we can retry but "
+						+ "the exception has been marked for rollback.", context.getLastThrowable());
+			}
+			return rollback;
+		}
+		// If no classifier is provided, just assume the all exceptions are for
+		// rollback if the execution is stateful, and none otherwise.
 		return state != null;
 	}
 

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/BinaryExceptionClassifier.java

@@ -20,59 +20,65 @@ import java.util.HashMap;
 import java.util.Map;
 
 /**
- * A {@link Classifier} that has only two classes of exception.
- * Provides convenient methods for setting up and querying the classification
- * with boolean return type.
+ * A {@link Classifier} for exceptions that has only two classes (true and
+ * false). Classifies objects according to their inheritance relation with the
+ * supplied types. If the object to be classified is one of the provided types,
+ * or is a subclass of one of the types, then the non-default value is returned
+ * (usually true).
+ * 
+ * @see SubclassClassifier
  * 
  * @author Dave Syer
  * 
  */
-public class BinaryExceptionClassifier extends ExceptionClassifierSupport {
+public class BinaryExceptionClassifier extends SubclassClassifier<Throwable, Boolean> {
 
 	/**
-	 * The classifier result for a non-default exception.
+	 * Create a binary exception classifier with the provided default value.
+	 * @param defaultValue
 	 */
-	public static final String NON_DEFAULT = "NON_DEFAULT";
-
-	private SubclassExceptionClassifier delegate = new SubclassExceptionClassifier();
+	public BinaryExceptionClassifier(boolean defaultValue) {
+		super(defaultValue);
+	}
 
 	/**
-	 * Set the special exceptions. Any exception on the list, or subclasses
-	 * thereof, will be classified as non-default.
+	 * Create a binary exception classifier with the default value false.
+	 */
+	public BinaryExceptionClassifier() {
+		this(false);
+	}
+
+	/**
+	 * Create a binary exception classifier with the provided classes and their
+	 * subclasses. The mapped value for these exceptions will be the one
+	 * provided (which will be the opposite of the default).
+	 * @param value
+	 */
+	public BinaryExceptionClassifier(Collection<Class<? extends Throwable>> exceptionClasses, boolean value) {
+		this(!value);
+		setTypes(exceptionClasses);
+	}
+
+	/**
+	 * Create a binary exception classifier with the default value false and
+	 * value mapping true for the provided classes and their subclasses.
+	 */
+	public BinaryExceptionClassifier(Collection<Class<? extends Throwable>> exceptionClasses) {
+		this(exceptionClasses, true);
+	}
+
+	/**
+	 * Set of Throwable class types to keys for the classifier. Any subclass of
+	 * the type provided will be classified as of non-default type.
 	 * 
-	 * @param exceptionClasses defaults to {@link Exception}.
+	 * @param types the types to classify as non-default
 	 */
-	public final void setExceptionClasses(Collection<Class<? extends Throwable>> exceptionClasses) {
-		Map<Class<? extends Throwable>, String> temp = new HashMap<Class<? extends Throwable>, String>();
-		for (Class<? extends Throwable> exceptionClass : exceptionClasses) {
-			temp.put(exceptionClass, NON_DEFAULT);
+	public final void setTypes(Collection<Class<? extends Throwable>> types) {
+		Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>();
+		for (Class<? extends Throwable> type : types) {
+			map.put(type, !getDefault());
 		}
-		this.delegate.setTypeMap(temp);
-	}
-
-	/**
-	 * Convenience method to return boolean if the throwable is classified as
-	 * default.
-	 * 
-	 * @param throwable the Throwable to classify
-	 * @return true if it is default classified (i.e. not on the list provided
-	 * in {@link #setExceptionClasses(Collection)}.
-	 */
-	public boolean isDefault(Throwable throwable) {
-		return classify(throwable).equals(DEFAULT);
-	}
-
-	/**
-	 * Returns either {@link ExceptionClassifierSupport#DEFAULT} or
-	 * {@link #NON_DEFAULT} depending on the type of the throwable. If the type
-	 * of the throwable or one of its ancestors is on the exception class list
-	 * the classification is as {@link #NON_DEFAULT}.
-	 * 
-	 * @see #setExceptionClasses(Collection)
-	 * @see ExceptionClassifierSupport#classify(Throwable)
-	 */
-	public String classify(Throwable throwable) {
-		return delegate.classify(throwable);
+		setTypeMap(map);
 	}
 
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/Classifier.java

@@ -26,20 +26,12 @@ package org.springframework.batch.support;
 public interface Classifier<C, T> {
 
 	/**
-	 * Get a default value, normally the same as would be returned by
-	 * {@link #classify(Object)} with null argument.
-	 * 
-	 * @return the default value.
-	 */
-	T getDefault();
-
-	/**
-	 * Classify the given object and return a non-null object. The return type
-	 * depends on the implementation but typically would be a key in a map which
-	 * the client maintains.
+	 * Classify the given object and return an object. The return type depends
+	 * on the implementation.
 	 * 
 	 * @param classifiable the input object. Can be null.
-	 * @return an object.
+	 * @return an object. Can be null, but implementations should declare if
+	 * this is the case.
 	 */
 	T classify(C classifiable);
 

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ClassifierSupport.java

@@ -17,35 +17,32 @@
 package org.springframework.batch.support;
 
 /**
- * Base class for {@link Classifier} implementations. Provides default
- * behaviour and some convenience members, like constants.
+ * Base class for {@link Classifier} implementations. Provides default behaviour
+ * and some convenience members, like constants.
  * 
  * @author Dave Syer
  * 
  */
-public class ExceptionClassifierSupport implements Classifier<Throwable,String> {
+public class ClassifierSupport<C, T> implements Classifier<C, T> {
+
+	final private T defaultValue;
 
 	/**
-	 * Default classification key.
+	 * @param defaultValue
 	 */
-	public static final String DEFAULT = "default";
+	public ClassifierSupport(T defaultValue) {
+		super();
+		this.defaultValue = defaultValue;
+	}
 
 	/**
-	 * Always returns the value of {@link #DEFAULT}.
+	 * Always returns the default value. This is the main extension point for
+	 * subclasses, so it must be able to classify null.
 	 * 
 	 * @see org.springframework.batch.support.Classifier#classify(Object)
 	 */
-	public String classify(Throwable throwable) {
-		return DEFAULT;
-	}
-
-	/**
-	 * Wrapper for a call to {@link #classify(Throwable)} with argument null.
-	 * 
-	 * @see org.springframework.batch.support.Classifier#getDefault()
-	 */
-	public String getDefault() {
-		return classify(null);
+	public T classify(C throwable) {
+		return defaultValue;
 	}
 
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SubclassClassifier.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.support;
+
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+/**
+ * A {@link Classifier} for a parameterised object type based on a map.
+ * Classifies objects according to their inheritance relation with the supplied
+ * type map. If the object to be classified is one of the keys of the provided
+ * map, or is a subclass of one of the keys, then the map entry vale for that
+ * key is returned. Otherwise returns the default value which is null by
+ * default.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class SubclassClassifier<T, C> implements Classifier<T, C> {
+
+	private Map<Class<? extends T>, C> classified = new HashMap<Class<? extends T>, C>();
+
+	private C defaultValue = null;
+
+	/**
+	 * Create a {@link SubclassClassifier} with null default value.
+	 * 
+	 */
+	public SubclassClassifier() {
+		this(null);
+	}
+
+	/**
+	 * Create a {@link SubclassClassifier} with supplied default value.
+	 * 
+	 * @param defaultValue
+	 */
+	public SubclassClassifier(C defaultValue) {
+		this(new HashMap<Class<? extends T>, C>(), defaultValue);
+	}
+
+	/**
+	 * Create a {@link SubclassClassifier} with supplied default value.
+	 * 
+	 * @param defaultValue
+	 */
+	public SubclassClassifier(Map<Class<? extends T>, C> typeMap, C defaultValue) {
+		super();
+		setTypeMap(typeMap);
+		this.defaultValue = defaultValue;
+	}
+
+	/**
+	 * Public setter for the default value for mapping keys that are not found
+	 * in the map (or their subclasses). Defaults to false.
+	 * 
+	 * @param defaultValue the default value to set
+	 */
+	public void setDefaultValue(C defaultValue) {
+		this.defaultValue = defaultValue;
+	}
+
+	/**
+	 * Set the classifications up as a map. The keys are types and these will be
+	 * mapped along with all their subclasses to the corresponding value. The
+	 * most specific types will match first.
+	 * 
+	 * @param map a map from type to class
+	 */
+	public void setTypeMap(Map<Class<? extends T>, C> map) {
+		this.classified = new HashMap<Class<? extends T>, C>(map);
+	}
+
+	/**
+	 * Return the value from the type map whose key is the class of the given
+	 * Throwable, or its nearest ancestor if a subclass.
+	 * 
+	 */
+	public C classify(T classifiable) {
+
+		if (classifiable == null) {
+			return defaultValue;
+		}
+
+		@SuppressWarnings("unchecked")
+		Class<? extends T> exceptionClass = (Class<? extends T>) classifiable.getClass();
+		if (classified.containsKey(exceptionClass)) {
+			return classified.get(exceptionClass);
+		}
+
+		// check for subclasses
+		Set<Class<? extends T>> classes = new TreeSet<Class<? extends T>>(new ClassComparator());
+		classes.addAll(classified.keySet());
+		for (Class<? extends T> cls : classes) {
+			if (cls.isAssignableFrom(exceptionClass)) {
+				C value = classified.get(cls);
+				this.classified.put(exceptionClass, value);
+				return value;
+			}
+		}
+
+		return defaultValue;
+	}
+
+	/**
+	 * Return the default value supplied in the constructor (default false).
+	 */
+	final public C getDefault() {
+		return defaultValue;
+	}
+
+	/**
+	 * Comparator for classes to order by inheritance.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
+	private static class ClassComparator implements Comparator<Class<?>> {
+		/**
+		 * @return 1 if arg0 is assignable from arg1, -1 otherwise
+		 * @see java.util.Comparator#compare(java.lang.Object, java.lang.Object)
+		 */
+		public int compare(Class<?> arg0, Class<?> arg1) {
+			if (arg0.isAssignableFrom(arg1)) {
+				return 1;
+			}
+			return -1;
+		}
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SubclassExceptionClassifier.java

@@ -1,105 +0,0 @@
-/*
- * Copyright 2006-2007 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.batch.support;
-
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-
-import org.springframework.util.Assert;
-
-/**
- * 
- * @author Dave Syer
- * 
- */
-public class SubclassExceptionClassifier extends ExceptionClassifierSupport {
-	
-	private Map<Class<? extends Throwable>, String> classified = new HashMap<Class<? extends Throwable>, String>();
-
-	/**
-	 * Map of Throwable class types to keys for the classifier. Any subclass of
-	 * the type provided will be classified as of the type given by the
-	 * corresponding map entry value.
-	 * 
-	 * @param typeMap the typeMap to set
-	 */
-	public final void setTypeMap(Map<Class<? extends Throwable>, String> typeMap) {
-		Map<Class<? extends Throwable>, String> map = new HashMap<Class<? extends Throwable>, String>();
-		for (Map.Entry<Class<? extends Throwable>, String> entry : typeMap.entrySet()) {
-			addRetryableExceptionClass(entry.getKey(), entry.getValue(), map);
-		}
-		this.classified = map;
-	}
-
-	/**
-	 * Return the value from the type map whose key is the class of the given
-	 * Throwable, or its nearest ancestor if a subclass.
-	 * 
-	 * @see org.springframework.batch.support.ExceptionClassifierSupport#classify(java.lang.Throwable)
-	 */
-	public String classify(Throwable throwable) {
-
-		if (throwable == null) {
-			return super.classify(throwable);
-		}
-
-		Class<? extends Throwable> exceptionClass = throwable.getClass();
-		if (classified.containsKey(exceptionClass)) {
-			return classified.get(exceptionClass);
-		}
-
-		// check for subclasses
-		Set<Class<? extends Throwable>> classes = new TreeSet<Class<? extends Throwable>>(new ClassComparator());
-		classes.addAll(classified.keySet());
-		for (Class<? extends Throwable> cls : classes) {
-			if (cls.isAssignableFrom(exceptionClass)) {
-				String value = classified.get(cls);
-				addRetryableExceptionClass(exceptionClass, value, this.classified);
-				return value;
-			}
-		}
-
-		return super.classify(throwable);
-	}
-
-	private void addRetryableExceptionClass(Class<? extends Throwable> exceptionClass, String classifiedAs, Map<Class<? extends Throwable>, String> map) {
-		Assert.isAssignable(Throwable.class, exceptionClass);
-		map.put(exceptionClass, classifiedAs);
-	}
-
-	/**
-	 * Comparator for classes to order by inheritance.
-	 * 
-	 * @author Dave Syer
-	 * 
-	 */
-	private class ClassComparator implements Comparator<Class<?>> {
-		/**
-		 * @return 1 if arg0 is assignable from arg1, -1 otherwise
-		 * @see java.util.Comparator#compare(java.lang.Object, java.lang.Object)
-		 */
-		public int compare(Class<?> arg0, Class<?> arg1) {
-			if (arg0.isAssignableFrom(arg1)) {
-				return 1;
-			}
-			return -1;
-		}
-	}
-
-}