From 8e498f0d5247bf7b9c004dbfbddd64557a03758d Mon Sep 17 00:00:00 2001
From: dsyer <dsyer@2e8e966d-6415-0410-94b9-dc4e65435d03>
Date: Sat, 23 Aug 2008 13:22:35 +0000
Subject: [PATCH] Improve parameterisation of exception handler

---
 .../item/SimpleRetryExceptionHandler.java     |    8 +-
 .../step/item/SkipLimitStepFactoryBean.java   | 1023 ++++++++---------
 .../skip/LimitCheckingItemSkipPolicy.java     |   16 +-
 .../LimitCheckingItemSkipPolicyTests.java     |    5 +-
 .../SimpleRetryExceptionHandlerTests.java     |   39 +-
 .../item/SkipLimitStepFactoryBeanTests.java   |   76 +-
 .../StatefulRetryStepFactoryBeanTests.java    |   53 +-
 .../LogOrRethrowExceptionHandler.java         |    6 +-
 .../RethrowOnThresholdExceptionHandler.java   |   18 +-
 .../SimpleLimitExceptionHandler.java          |    2 +-
 .../ExceptionClassifierRetryPolicy.java       |    8 +-
 .../batch/retry/policy/SimpleRetryPolicy.java |   16 +-
 .../support/BinaryExceptionClassifier.java    |   11 +-
 .../batch/support/ExceptionClassifier.java    |   10 +-
 .../support/ExceptionClassifierSupport.java   |    4 +-
 .../support/SubclassExceptionClassifier.java  |   16 +-
 ...throwOnThresholdExceptionHandlerTests.java |   19 +-
 .../retry/policy/SimpleRetryPolicyTests.java  |   31 +-
 .../retry/support/RetryTemplateTests.java     |    8 +-
 .../BinaryExceptionClassifierTests.java       |    9 +-
 .../SubclassExceptionClassifierTests.java     |    8 +-
 21 files changed, 731 insertions(+), 655 deletions(-)

diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandler.java
index 2637ea000..55dbb84cc 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandler.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandler.java
@@ -15,6 +15,8 @@
  */
 package org.springframework.batch.core.step.item;
 
+import java.util.Collection;
+
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.exception.ExceptionHandler;
 import org.springframework.batch.repeat.support.RepeatSynchronizationManager;
@@ -52,13 +54,13 @@ public class SimpleRetryExceptionHandler extends RetryListenerSupport implements
 	 * exception is encountered
 	 * @param exceptionHandler the delegate to use if an exception actually
 	 * needs to be handled
-	 * @param classes
+	 * @param fatalExceptionClasses
 	 */
-	public SimpleRetryExceptionHandler(RetryPolicy retryPolicy, ExceptionHandler exceptionHandler, Class<?>[] classes) {
+	public SimpleRetryExceptionHandler(RetryPolicy retryPolicy, ExceptionHandler exceptionHandler, Collection<Class<? extends Throwable>> fatalExceptionClasses) {
 		this.retryPolicy = retryPolicy;
 		this.exceptionHandler = exceptionHandler;
 		this.fatalExceptionClassifier = new BinaryExceptionClassifier();
-		fatalExceptionClassifier.setExceptionClasses(classes);
+		fatalExceptionClassifier.setExceptionClasses(fatalExceptionClasses);
 	}
 
 	/**
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
index 70bed93cf..c6c2ed351 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
@@ -1,517 +1,506 @@
-package org.springframework.batch.core.step.item;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-
-import org.springframework.batch.core.SkipListener;
-import org.springframework.batch.core.StepContribution;
-import org.springframework.batch.core.listener.CompositeSkipListener;
-import org.springframework.batch.core.step.skip.ItemSkipPolicy;
-import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
-import org.springframework.batch.core.step.skip.SkipLimitExceededException;
-import org.springframework.batch.core.step.skip.SkipListenerFailedException;
-import org.springframework.batch.item.ItemKeyGenerator;
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.item.ItemReader;
-import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.repeat.RepeatOperations;
-import org.springframework.batch.repeat.support.RepeatTemplate;
-import org.springframework.batch.retry.RecoveryCallback;
-import org.springframework.batch.retry.RetryCallback;
-import org.springframework.batch.retry.RetryContext;
-import org.springframework.batch.retry.RetryException;
-import org.springframework.batch.retry.RetryListener;
-import org.springframework.batch.retry.RetryOperations;
-import org.springframework.batch.retry.RetryPolicy;
-import org.springframework.batch.retry.backoff.BackOffPolicy;
-import org.springframework.batch.retry.callback.RecoveryRetryCallback;
-import org.springframework.batch.retry.policy.ExceptionClassifierRetryPolicy;
-import org.springframework.batch.retry.policy.MapRetryContextCache;
-import org.springframework.batch.retry.policy.NeverRetryPolicy;
-import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
-import org.springframework.batch.retry.policy.RetryContextCache;
-import org.springframework.batch.retry.policy.SimpleRetryPolicy;
-import org.springframework.batch.retry.support.RetryTemplate;
-import org.springframework.batch.support.SubclassExceptionClassifier;
-
-/**
- * Factory bean for step that provides options for configuring skip behavior.
- * User can set {@link #setSkipLimit(int)} to set how many exceptions of
- * {@link #setSkippableExceptionClasses(Class[])} types are tolerated.
- * {@link #setFatalExceptionClasses(Class[])} will cause immediate termination
- * of job - they are treated as higher priority than
- * {@link #setSkippableExceptionClasses(Class[])}, so the two lists don't need
- * to be exclusive.
- * 
- * Skippable exceptions on write will by default cause transaction rollback - to
- * avoid rollback for specific exception class include it in the transaction
- * attribute as "no rollback for".
- * 
- * @see SimpleStepFactoryBean
- * 
- * @author Dave Syer
- * @author Robert Kasanicky
- * 
- */
-public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S> {
-
-	private int skipLimit = 0;
-
-	private Class<?>[] skippableExceptionClasses = new Class<?>[] { Exception.class };
-
-	private Class<?>[] fatalExceptionClasses = new Class<?>[] { Error.class };
-
-	private ItemKeyGenerator itemKeyGenerator;
-
-	private int cacheCapacity = 0;
-
-	private int retryLimit = 0;
-
-	private Class<?>[] retryableExceptionClasses = new Class<?>[] {};
-
-	private BackOffPolicy backOffPolicy;
-
-	private RetryListener[] retryListeners;
-
-	private RetryPolicy retryPolicy;
-
-	private RetryContextCache retryContextCache;
-
-	/**
-	 * Setter for the retry policy. If this is specified the other retry
-	 * properties are ignored (retryLimit, backOffPolicy,
-	 * retryableExceptionClasses).
-	 * 
-	 * @param retryPolicy a stateless {@link RetryPolicy}
-	 */
-	public void setRetryPolicy(RetryPolicy retryPolicy) {
-		this.retryPolicy = retryPolicy;
-	}
-
-	/**
-	 * Public setter for the retry limit. Each item can be retried up to this
-	 * limit.
-	 * @param retryLimit the retry limit to set
-	 */
-	public void setRetryLimit(int retryLimit) {
-		this.retryLimit = retryLimit;
-	}
-
-	/**
-	 * Public setter for the capacity of the cache in the retry policy. If more
-	 * items than this fail without being skipped or recovered an exception will
-	 * be thrown. This is to guard against inadvertent infinite loops generated
-	 * by item identity problems.<br/>
-	 * 
-	 * The default value should be high enough and more for most purposes. To
-	 * breach the limit in a single-threaded step typically you have to have
-	 * this many failures in a single transaction. Defaults to the value in the
-	 * {@link MapRetryContextCache}.<br/>
-	 * 
-	 * This property is ignored if the
-	 * {@link #setRetryContextCache(RetryContextCache)} is set directly.
-	 * 
-	 * @param cacheCapacity the cache capacity to set (greater than 0 else
-	 * ignored)
-	 */
-	public void setCacheCapacity(int cacheCapacity) {
-		this.cacheCapacity = cacheCapacity;
-	}
-
-	/**
-	 * Override the default retry context cache for retry of chunk processing.
-	 * If this property is set then {@link #setCacheCapacity(int)} is ignored.
-	 * 
-	 * @param retryContextCache the {@link RetryContextCache} to set
-	 */
-	public void setRetryContextCache(RetryContextCache retryContextCache) {
-		this.retryContextCache = retryContextCache;
-	}
-
-	/**
-	 * Public setter for the Class[].
-	 * @param retryableExceptionClasses the retryableExceptionClasses to set
-	 */
-	public void setRetryableExceptionClasses(Class<?>[] retryableExceptionClasses) {
-		this.retryableExceptionClasses = retryableExceptionClasses;
-	}
-
-	/**
-	 * Public setter for the {@link BackOffPolicy}.
-	 * @param backOffPolicy the {@link BackOffPolicy} to set
-	 */
-	public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
-		this.backOffPolicy = backOffPolicy;
-	}
-
-	/**
-	 * Public setter for the {@link RetryListener}s.
-	 * @param retryListeners the {@link RetryListener}s to set
-	 */
-	public void setRetryListeners(RetryListener[] retryListeners) {
-		this.retryListeners = retryListeners;
-	}
-
-	/**
-	 * 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
-	 * skipped and no exception propagated until the limit is reached. If it is
-	 * zero then all exceptions will be propagated from the chunk and cause the
-	 * step to abort.
-	 * 
-	 * @param skipLimit the value to set. Default is 0 (never skip).
-	 */
-	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.
-	 * 
-	 * @param exceptionClasses defaults to <code>Exception</code>
-	 */
-	public void setSkippableExceptionClasses(Class<?>[] exceptionClasses) {
-		this.skippableExceptionClasses = exceptionClasses;
-	}
-
-	/**
-	 * Public setter for exception classes that should cause immediate failure.
-	 * 
-	 * @param fatalExceptionClasses {@link Error} by default
-	 */
-	public void setFatalExceptionClasses(Class<?>[] fatalExceptionClasses) {
-		this.fatalExceptionClasses = fatalExceptionClasses;
-	}
-
-	/**
-	 * Public setter for the {@link ItemKeyGenerator}. This is used to identify
-	 * failed items so they can be skipped if encountered again, generally in
-	 * another transaction.
-	 * 
-	 * @param itemKeyGenerator the {@link ItemKeyGenerator} to set.
-	 */
-	public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
-		this.itemKeyGenerator = itemKeyGenerator;
-	}
-
-	/**
-	 * Uses the {@link #setSkipLimit(int)} value to configure item handler and
-	 * and exception handler.
-	 */
-	protected void applyConfiguration(StepHandlerStep step) {
-		super.applyConfiguration(step);
-
-		if (retryLimit > 0 || skipLimit > 0 || retryPolicy != null) {
-
-			addFatalExceptionIfMissing(SkipLimitExceededException.class);
-			addFatalExceptionIfMissing(SkipListenerFailedException.class);
-			addFatalExceptionIfMissing(RetryException.class);
-
-			if (retryPolicy == null) {
-
-				SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
-				if (retryableExceptionClasses.length > 0) { // otherwise we
-					// retry
-					// all exceptions
-					simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
-				}
-				simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses);
-
-				ExceptionClassifierRetryPolicy classifierRetryPolicy = new ExceptionClassifierRetryPolicy();
-				SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
-				HashMap<Class<?>, String> exceptionTypeMap = new HashMap<Class<?>, String>();
-				for (Class<?> cls : retryableExceptionClasses) {
-					exceptionTypeMap.put(cls, "retry");
-				}
-				exceptionClassifier.setTypeMap(exceptionTypeMap);
-				HashMap<String, RetryPolicy> retryPolicyMap = new HashMap<String, RetryPolicy>();
-				retryPolicyMap.put("retry", simpleRetryPolicy);
-				retryPolicyMap.put("default", new NeverRetryPolicy());
-				classifierRetryPolicy.setPolicyMap(retryPolicyMap);
-				classifierRetryPolicy.setExceptionClassifier(exceptionClassifier);
-				retryPolicy = classifierRetryPolicy;
-
-			}
-
-			// Co-ordinate the retry policy with the exception handler:
-			RepeatOperations stepOperations = getStepOperations();
-			if (stepOperations instanceof RepeatTemplate) {
-				((RepeatTemplate) stepOperations).setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy,
-						getExceptionHandler(), fatalExceptionClasses));
-			}
-
-			RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy) {
-				protected boolean recoverForException(Throwable ex) {
-					return !getTransactionAttribute().rollbackOn(ex);
-				}
-			};
-
-			if (retryContextCache == null) {
-				if (cacheCapacity > 0) {
-					recoveryCallbackRetryPolicy.setRetryContextCache(new MapRetryContextCache(cacheCapacity));
-				}
-			}
-			else {
-				recoveryCallbackRetryPolicy.setRetryContextCache(retryContextCache);
-			}
-
-			RetryTemplate retryTemplate = new RetryTemplate();
-			if (retryListeners != null) {
-				retryTemplate.setListeners(retryListeners);
-			}
-			retryTemplate.setRetryPolicy(recoveryCallbackRetryPolicy);
-			if (retryPolicy == null && backOffPolicy != null) {
-				retryTemplate.setBackOffPolicy(backOffPolicy);
-			}
-
-			List<Class<?>> exceptions = new ArrayList<Class<?>>(Arrays.asList(skippableExceptionClasses));
-			ItemSkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
-					new ArrayList<Class<?>>() {
-						{
-							for (Class<?> exceptionClass : fatalExceptionClasses) {
-								add(exceptionClass);
-							}
-						}
-					});
-			exceptions.addAll(new ArrayList<Class<?>>() {
-				{
-					for (Class<?> exceptionClass : retryableExceptionClasses) {
-						add(exceptionClass);
-					}
-				}
-			});
-			ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
-					new ArrayList<Class<?>>() {
-						{
-							for (Class<?> exceptionClass : fatalExceptionClasses) {
-								add(exceptionClass);
-							}
-						}
-					});
-			StatefulRetryStepHandler<T, S> itemHandler = new StatefulRetryStepHandler<T, S>(getItemReader(),
-					getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator,
-					readSkipPolicy, writeSkipPolicy);
-			itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners()));
-
-			step.setStepHandler(itemHandler);
-
-		}
-
-	}
-
-	public void addFatalExceptionIfMissing(Class<?> cls) {
-		List<Class<?>> fatalExceptionList = new ArrayList<Class<?>>();
-		for (Class<?> exceptionClass : fatalExceptionClasses) {
-			fatalExceptionList.add(exceptionClass);
-		}
-		if (!fatalExceptionList.contains(cls)) {
-			fatalExceptionList.add(cls);
-		}
-		fatalExceptionClasses = fatalExceptionList.toArray(new Class[0]);
-	}
-
-	/**
-	 * If there is an exception on input it is skipped if allowed. If there is
-	 * an exception on output, it will be re-thrown in any case, and the
-	 * behaviour when the item is next encountered depends on the retryable and
-	 * skippable exception configuration. If the exception is retryable the
-	 * write will be attempted again up to the retry limit. When retry attempts
-	 * are exhausted the skip listener is invoked and the skip count
-	 * incremented. A retryable exception is thus also effectively also
-	 * implicitly skippable.
-	 * 
-	 * @author Dave Syer
-	 * 
-	 */
-	private static class StatefulRetryStepHandler<T, S> extends ItemOrientedStepHandler<T, S> {
-
-		final private RetryOperations retryOperations;
-
-		final private ItemKeyGenerator itemKeyGenerator;
-
-		final private CompositeSkipListener listener = new CompositeSkipListener();
-
-		final private ItemSkipPolicy readSkipPolicy;
-
-		final private ItemSkipPolicy writeSkipPolicy;
-
-		/**
-		 * @param itemReader
-		 * @param itemWriter
-		 * @param retryTemplate
-		 * @param itemKeyGenerator
-		 */
-		public StatefulRetryStepHandler(ItemReader<? extends T> itemReader,
-				ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
-				RepeatOperations chunkOperations, RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator,
-				ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) {
-			super(itemReader, itemProcessor, itemWriter, chunkOperations);
-			this.retryOperations = retryTemplate;
-			this.itemKeyGenerator = itemKeyGenerator;
-			this.readSkipPolicy = readSkipPolicy;
-			this.writeSkipPolicy = writeSkipPolicy;
-		}
-
-		/**
-		 * Register some {@link SkipListener}s with the handler. Each will get
-		 * the callbacks in the order specified at the correct stage if a skip
-		 * occurs.
-		 * 
-		 * @param listeners
-		 */
-		public void setSkipListeners(SkipListener[] listeners) {
-			for (SkipListener listener : listeners) {
-				registerSkipListener(listener);
-			}
-		}
-
-		/**
-		 * Register a listener for callbacks at the appropriate stages in a skip
-		 * process.
-		 * 
-		 * @param listener a {@link SkipListener}
-		 */
-		public void registerSkipListener(SkipListener listener) {
-			this.listener.register(listener);
-		}
-
-		/**
-		 * Tries to read the item from the reader, in case of exception skip the
-		 * item if the skip policy allows, otherwise re-throw.
-		 * 
-		 * @param contribution current StepContribution holding skipped items
-		 * count
-		 * @return next item for processing
-		 */
-		protected ItemWrapper<T> read(StepContribution contribution) throws Exception {
-
-			int skipCount = 0;
-
-			while (true) {
-				try {
-					return new ItemWrapper<T>(doRead(), skipCount);
-				}
-				catch (Exception e) {
-					try {
-						if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
-							// increment skip count and try again
-							try {
-								skipCount++;
-								listener.onSkipInRead(e);
-							}
-							catch (RuntimeException ex) {
-								contribution.incrementReadSkipCount(skipCount);
-								throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
-							}
-							logger.debug("Skipping failed input", e);
-						}
-						else {
-							// re-throw only when the skip policy runs out of
-							// patience
-							throw e;
-						}
-					}
-					catch (SkipLimitExceededException ex) {
-						// we are headed for a abnormal ending so bake in the
-						// skip count
-						contribution.incrementReadSkipCount(skipCount);
-						throw ex;
-					}
-				}
-			}
-
-		}
-
-		/**
-		 * Execute the business logic, delegating to the writer.<br/>
-		 * 
-		 * Process the items with the {@link ItemWriter} in a stateful retry.
-		 * Any {@link SkipListener} provided is called when retry attempts are
-		 * exhausted. The listener callback (on write failure) will happen in
-		 * the next transaction automatically.<br/>
-		 */
-		@Override
-		protected void write(final Chunk<S> chunk, StepContribution contribution) throws Exception {
-
-			if (!chunk.canRetry()) {
-				logger.debug("Run items: " + chunk.getItems());
-				runChunk(chunk, contribution);
-			}
-			else {
-				logger.debug(String.format("Retry items: %s", chunk.getItems()));
-				retryChunk(chunk, contribution);
-			}
-			chunk.rethrow();
-
-			chunk.clear();
-
-		}
-
-		/**
-		 * @param chunk
-		 */
-		private void runChunk(Chunk<S> chunk, final StepContribution contribution) throws Exception {
-			try {
-				doWrite(chunk.getItems());
-			}
-			catch (Exception e) {
-				chunk.rethrow(e);
-			}
-		}
-
-		/**
-		 * @param chunk
-		 */
-		private void retryChunk(final Chunk<S> chunk, final StepContribution contribution) throws Exception {
-
-			RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(chunk, new RetryCallback() {
-				public Object doWithRetry(RetryContext context) throws Throwable {
-					doWrite(chunk.getItems());
-					return null;
-				}
-			}, chunk);
-
-			retryCallback.setRecoveryCallback(new RecoveryCallback() {
-
-				public Object recover(RetryContext context) throws Exception {
-
-					Exception t = (Exception) context.getLastThrowable();
-
-					if (!chunk.canSkip()) {
-						throw t;
-					}
-
-					if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
-						contribution.incrementWriteSkipCount();
-						S item = chunk.getSkippedItem();
-						try {
-							listener.onSkipInWrite(item, t);
-							return null;
-						}
-						catch (RuntimeException ex) {
-							throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
-						}
-					}
-					else {
-						throw new RetryException("Non-skippable exception in recoverer", t);
-					}
-
-				}
-			});
-
-			try {
-				retryOperations.execute(retryCallback);
-			}
-			catch (Exception e) {
-				// only if the retry failed do we re-arrange the chunk
-				chunk.rethrow(e);
-			}
-
-		}
-	}
-
-}
+package org.springframework.batch.core.step.item;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+
+import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.listener.CompositeSkipListener;
+import org.springframework.batch.core.step.skip.ItemSkipPolicy;
+import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
+import org.springframework.batch.core.step.skip.SkipLimitExceededException;
+import org.springframework.batch.core.step.skip.SkipListenerFailedException;
+import org.springframework.batch.item.ItemKeyGenerator;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.repeat.RepeatOperations;
+import org.springframework.batch.repeat.support.RepeatTemplate;
+import org.springframework.batch.retry.RecoveryCallback;
+import org.springframework.batch.retry.RetryCallback;
+import org.springframework.batch.retry.RetryContext;
+import org.springframework.batch.retry.RetryException;
+import org.springframework.batch.retry.RetryListener;
+import org.springframework.batch.retry.RetryOperations;
+import org.springframework.batch.retry.RetryPolicy;
+import org.springframework.batch.retry.backoff.BackOffPolicy;
+import org.springframework.batch.retry.callback.RecoveryRetryCallback;
+import org.springframework.batch.retry.policy.ExceptionClassifierRetryPolicy;
+import org.springframework.batch.retry.policy.MapRetryContextCache;
+import org.springframework.batch.retry.policy.NeverRetryPolicy;
+import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
+import org.springframework.batch.retry.policy.RetryContextCache;
+import org.springframework.batch.retry.policy.SimpleRetryPolicy;
+import org.springframework.batch.retry.support.RetryTemplate;
+import org.springframework.batch.support.SubclassExceptionClassifier;
+
+/**
+ * Factory bean for step that provides options for configuring skip behaviour.
+ * User can set {@link #setSkipLimit(int)} to set how many exceptions of
+ * {@link #setSkippableExceptionClasses(Collection)} types are tolerated.
+ * {@link #setFatalExceptionClasses(Collection)} will cause immediate
+ * termination of job - they are treated as higher priority than
+ * {@link #setSkippableExceptionClasses(Collection)}, so the two lists don't
+ * need to be exclusive.
+ * 
+ * Skippable exceptions on write will by default cause transaction rollback - to
+ * avoid rollback for specific exception class include it in the transaction
+ * attribute as "no rollback for".
+ * 
+ * @see SimpleStepFactoryBean
+ * 
+ * @author Dave Syer
+ * @author Robert Kasanicky
+ * 
+ */
+public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S> {
+
+	private int skipLimit = 0;
+
+	private Collection<Class<? extends Throwable>> skippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
+
+	private Collection<Class<? extends Throwable>> fatalExceptionClasses = new HashSet<Class<? extends Throwable>>();
+
+	{
+		fatalExceptionClasses.add(Error.class);
+		skippableExceptionClasses.add(Exception.class);
+	}
+
+	private ItemKeyGenerator itemKeyGenerator;
+
+	private int cacheCapacity = 0;
+
+	private int retryLimit = 0;
+
+	private Collection<Class<? extends Throwable>> retryableExceptionClasses = new HashSet<Class<? extends Throwable>>();
+
+	private BackOffPolicy backOffPolicy;
+
+	private RetryListener[] retryListeners;
+
+	private RetryPolicy retryPolicy;
+
+	private RetryContextCache retryContextCache;
+
+	/**
+	 * Setter for the retry policy. If this is specified the other retry
+	 * properties are ignored (retryLimit, backOffPolicy,
+	 * retryableExceptionClasses).
+	 * 
+	 * @param retryPolicy a stateless {@link RetryPolicy}
+	 */
+	public void setRetryPolicy(RetryPolicy retryPolicy) {
+		this.retryPolicy = retryPolicy;
+	}
+
+	/**
+	 * Public setter for the retry limit. Each item can be retried up to this
+	 * limit.
+	 * @param retryLimit the retry limit to set
+	 */
+	public void setRetryLimit(int retryLimit) {
+		this.retryLimit = retryLimit;
+	}
+
+	/**
+	 * Public setter for the capacity of the cache in the retry policy. If more
+	 * items than this fail without being skipped or recovered an exception will
+	 * be thrown. This is to guard against inadvertent infinite loops generated
+	 * by item identity problems.<br/>
+	 * 
+	 * The default value should be high enough and more for most purposes. To
+	 * breach the limit in a single-threaded step typically you have to have
+	 * this many failures in a single transaction. Defaults to the value in the
+	 * {@link MapRetryContextCache}.<br/>
+	 * 
+	 * This property is ignored if the
+	 * {@link #setRetryContextCache(RetryContextCache)} is set directly.
+	 * 
+	 * @param cacheCapacity the cache capacity to set (greater than 0 else
+	 * ignored)
+	 */
+	public void setCacheCapacity(int cacheCapacity) {
+		this.cacheCapacity = cacheCapacity;
+	}
+
+	/**
+	 * Override the default retry context cache for retry of chunk processing.
+	 * If this property is set then {@link #setCacheCapacity(int)} is ignored.
+	 * 
+	 * @param retryContextCache the {@link RetryContextCache} to set
+	 */
+	public void setRetryContextCache(RetryContextCache retryContextCache) {
+		this.retryContextCache = retryContextCache;
+	}
+
+	/**
+	 * Public setter for the Class[].
+	 * @param retryableExceptionClasses the retryableExceptionClasses to set
+	 */
+	public void setRetryableExceptionClasses(Collection<Class<? extends Throwable>> retryableExceptionClasses) {
+		this.retryableExceptionClasses = retryableExceptionClasses;
+	}
+
+	/**
+	 * Public setter for the {@link BackOffPolicy}.
+	 * @param backOffPolicy the {@link BackOffPolicy} to set
+	 */
+	public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
+		this.backOffPolicy = backOffPolicy;
+	}
+
+	/**
+	 * Public setter for the {@link RetryListener}s.
+	 * @param retryListeners the {@link RetryListener}s to set
+	 */
+	public void setRetryListeners(RetryListener[] retryListeners) {
+		this.retryListeners = retryListeners;
+	}
+
+	/**
+	 * 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
+	 * skipped and no exception propagated until the limit is reached. If it is
+	 * zero then all exceptions will be propagated from the chunk and cause the
+	 * step to abort.
+	 * 
+	 * @param skipLimit the value to set. Default is 0 (never skip).
+	 */
+	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.
+	 * 
+	 * @param exceptionClasses defaults to <code>Exception</code>
+	 */
+	public void setSkippableExceptionClasses(Collection<Class<? extends Throwable>> exceptionClasses) {
+		this.skippableExceptionClasses = exceptionClasses;
+	}
+
+	/**
+	 * Public setter for exception classes that should cause immediate failure.
+	 * 
+	 * @param fatalExceptionClasses {@link Error} by default
+	 */
+	public void setFatalExceptionClasses(Collection<Class<? extends Throwable>> fatalExceptionClasses) {
+		this.fatalExceptionClasses = fatalExceptionClasses;
+	}
+
+	/**
+	 * Public setter for the {@link ItemKeyGenerator}. This is used to identify
+	 * failed items so they can be skipped if encountered again, generally in
+	 * another transaction.
+	 * 
+	 * @param itemKeyGenerator the {@link ItemKeyGenerator} to set.
+	 */
+	public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
+		this.itemKeyGenerator = itemKeyGenerator;
+	}
+
+	/**
+	 * Uses the {@link #setSkipLimit(int)} value to configure item handler and
+	 * and exception handler.
+	 */
+	protected void applyConfiguration(StepHandlerStep step) {
+		super.applyConfiguration(step);
+
+		if (retryLimit > 0 || skipLimit > 0 || retryPolicy != null) {
+
+			addFatalExceptionIfMissing(SkipLimitExceededException.class);
+			addFatalExceptionIfMissing(SkipListenerFailedException.class);
+			addFatalExceptionIfMissing(RetryException.class);
+
+			if (retryPolicy == null) {
+
+				SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
+				if (!retryableExceptionClasses.isEmpty()) { // otherwise we
+					// retry
+					// all exceptions
+					simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
+				}
+				simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses);
+
+				ExceptionClassifierRetryPolicy classifierRetryPolicy = new ExceptionClassifierRetryPolicy();
+				SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
+				HashMap<Class<? extends Throwable>, String> exceptionTypeMap = new HashMap<Class<? extends Throwable>, String>();
+				for (Class<? extends Throwable> cls : retryableExceptionClasses) {
+					exceptionTypeMap.put(cls, "retry");
+				}
+				exceptionClassifier.setTypeMap(exceptionTypeMap);
+				HashMap<String, RetryPolicy> retryPolicyMap = new HashMap<String, RetryPolicy>();
+				retryPolicyMap.put("retry", simpleRetryPolicy);
+				retryPolicyMap.put("default", new NeverRetryPolicy());
+				classifierRetryPolicy.setPolicyMap(retryPolicyMap);
+				classifierRetryPolicy.setExceptionClassifier(exceptionClassifier);
+				retryPolicy = classifierRetryPolicy;
+
+			}
+
+			// Co-ordinate the retry policy with the exception handler:
+			RepeatOperations stepOperations = getStepOperations();
+			if (stepOperations instanceof RepeatTemplate) {
+				((RepeatTemplate) stepOperations).setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy,
+						getExceptionHandler(), fatalExceptionClasses));
+			}
+
+			RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy) {
+				protected boolean recoverForException(Throwable ex) {
+					return !getTransactionAttribute().rollbackOn(ex);
+				}
+			};
+
+			if (retryContextCache == null) {
+				if (cacheCapacity > 0) {
+					recoveryCallbackRetryPolicy.setRetryContextCache(new MapRetryContextCache(cacheCapacity));
+				}
+			}
+			else {
+				recoveryCallbackRetryPolicy.setRetryContextCache(retryContextCache);
+			}
+
+			RetryTemplate retryTemplate = new RetryTemplate();
+			if (retryListeners != null) {
+				retryTemplate.setListeners(retryListeners);
+			}
+			retryTemplate.setRetryPolicy(recoveryCallbackRetryPolicy);
+			if (retryPolicy == null && backOffPolicy != null) {
+				retryTemplate.setBackOffPolicy(backOffPolicy);
+			}
+
+			List<Class<? extends Throwable>> exceptions = new ArrayList<Class<? extends Throwable>>(
+					skippableExceptionClasses);
+			ItemSkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses,
+					new ArrayList<Class<? extends Throwable>>(fatalExceptionClasses));
+			exceptions.addAll(new ArrayList<Class<? extends Throwable>>(retryableExceptionClasses));
+			ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
+					new ArrayList<Class<? extends Throwable>>(fatalExceptionClasses));
+			StatefulRetryStepHandler<T, S> itemHandler = new StatefulRetryStepHandler<T, S>(getItemReader(),
+					getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator,
+					readSkipPolicy, writeSkipPolicy);
+			itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners()));
+
+			step.setStepHandler(itemHandler);
+
+		}
+
+	}
+
+	public void addFatalExceptionIfMissing(Class<? extends Throwable> cls) {
+		List<Class<? extends Throwable>> fatalExceptionList = new ArrayList<Class<? extends Throwable>>();
+		for (Class<? extends Throwable> exceptionClass : fatalExceptionClasses) {
+			fatalExceptionList.add(exceptionClass);
+		}
+		if (!fatalExceptionList.contains(cls)) {
+			fatalExceptionList.add(cls);
+		}
+		fatalExceptionClasses = fatalExceptionList;
+	}
+
+	/**
+	 * If there is an exception on input it is skipped if allowed. If there is
+	 * an exception on output, it will be re-thrown in any case, and the
+	 * behaviour when the item is next encountered depends on the retryable and
+	 * skippable exception configuration. If the exception is retryable the
+	 * write will be attempted again up to the retry limit. When retry attempts
+	 * are exhausted the skip listener is invoked and the skip count
+	 * incremented. A retryable exception is thus also effectively also
+	 * implicitly skippable.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 */
+	private static class StatefulRetryStepHandler<T, S> extends ItemOrientedStepHandler<T, S> {
+
+		final private RetryOperations retryOperations;
+
+		final private ItemKeyGenerator itemKeyGenerator;
+
+		final private CompositeSkipListener listener = new CompositeSkipListener();
+
+		final private ItemSkipPolicy readSkipPolicy;
+
+		final private ItemSkipPolicy writeSkipPolicy;
+
+		/**
+		 * @param itemReader
+		 * @param itemWriter
+		 * @param retryTemplate
+		 * @param itemKeyGenerator
+		 */
+		public StatefulRetryStepHandler(ItemReader<? extends T> itemReader,
+				ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
+				RepeatOperations chunkOperations, RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator,
+				ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) {
+			super(itemReader, itemProcessor, itemWriter, chunkOperations);
+			this.retryOperations = retryTemplate;
+			this.itemKeyGenerator = itemKeyGenerator;
+			this.readSkipPolicy = readSkipPolicy;
+			this.writeSkipPolicy = writeSkipPolicy;
+		}
+
+		/**
+		 * Register some {@link SkipListener}s with the handler. Each will get
+		 * the callbacks in the order specified at the correct stage if a skip
+		 * occurs.
+		 * 
+		 * @param listeners
+		 */
+		public void setSkipListeners(SkipListener[] listeners) {
+			for (SkipListener listener : listeners) {
+				registerSkipListener(listener);
+			}
+		}
+
+		/**
+		 * Register a listener for callbacks at the appropriate stages in a skip
+		 * process.
+		 * 
+		 * @param listener a {@link SkipListener}
+		 */
+		public void registerSkipListener(SkipListener listener) {
+			this.listener.register(listener);
+		}
+
+		/**
+		 * Tries to read the item from the reader, in case of exception skip the
+		 * item if the skip policy allows, otherwise re-throw.
+		 * 
+		 * @param contribution current StepContribution holding skipped items
+		 * count
+		 * @return next item for processing
+		 */
+		protected ItemWrapper<T> read(StepContribution contribution) throws Exception {
+
+			int skipCount = 0;
+
+			while (true) {
+				try {
+					return new ItemWrapper<T>(doRead(), skipCount);
+				}
+				catch (Exception e) {
+					try {
+						if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+							// increment skip count and try again
+							try {
+								skipCount++;
+								listener.onSkipInRead(e);
+							}
+							catch (RuntimeException ex) {
+								contribution.incrementReadSkipCount(skipCount);
+								throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
+							}
+							logger.debug("Skipping failed input", e);
+						}
+						else {
+							// re-throw only when the skip policy runs out of
+							// patience
+							throw e;
+						}
+					}
+					catch (SkipLimitExceededException ex) {
+						// we are headed for a abnormal ending so bake in the
+						// skip count
+						contribution.incrementReadSkipCount(skipCount);
+						throw ex;
+					}
+				}
+			}
+
+		}
+
+		/**
+		 * Execute the business logic, delegating to the writer.<br/>
+		 * 
+		 * Process the items with the {@link ItemWriter} in a stateful retry.
+		 * Any {@link SkipListener} provided is called when retry attempts are
+		 * exhausted. The listener callback (on write failure) will happen in
+		 * the next transaction automatically.<br/>
+		 */
+		@Override
+		protected void write(final Chunk<S> chunk, StepContribution contribution) throws Exception {
+
+			if (!chunk.canRetry()) {
+				logger.debug("Run items: " + chunk.getItems());
+				runChunk(chunk, contribution);
+			}
+			else {
+				logger.debug(String.format("Retry items: %s", chunk.getItems()));
+				retryChunk(chunk, contribution);
+			}
+			chunk.rethrow();
+
+			chunk.clear();
+
+		}
+
+		/**
+		 * @param chunk
+		 */
+		private void runChunk(Chunk<S> chunk, final StepContribution contribution) throws Exception {
+			try {
+				doWrite(chunk.getItems());
+			}
+			catch (Exception e) {
+				chunk.rethrow(e);
+			}
+		}
+
+		/**
+		 * @param chunk
+		 */
+		private void retryChunk(final Chunk<S> chunk, final StepContribution contribution) throws Exception {
+
+			RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(chunk, new RetryCallback() {
+				public Object doWithRetry(RetryContext context) throws Throwable {
+					doWrite(chunk.getItems());
+					return null;
+				}
+			}, chunk);
+
+			retryCallback.setRecoveryCallback(new RecoveryCallback() {
+
+				public Object recover(RetryContext context) throws Exception {
+
+					Exception t = (Exception) context.getLastThrowable();
+
+					if (!chunk.canSkip()) {
+						throw t;
+					}
+
+					if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
+						contribution.incrementWriteSkipCount();
+						S item = chunk.getSkippedItem();
+						try {
+							listener.onSkipInWrite(item, t);
+							return null;
+						}
+						catch (RuntimeException ex) {
+							throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
+						}
+					}
+					else {
+						throw new RetryException("Non-skippable exception in recoverer", t);
+					}
+
+				}
+			});
+
+			try {
+				retryOperations.execute(retryCallback);
+			}
+			catch (Exception e) {
+				// only if the retry failed do we re-arrange the chunk
+				chunk.rethrow(e);
+			}
+
+		}
+	}
+
+}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java
index e1056e59f..11e0736ce 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java
@@ -16,11 +16,11 @@
 package org.springframework.batch.core.step.skip;
 
 import java.io.FileNotFoundException;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.List;
+import java.util.HashSet;
 import java.util.Map;
-import java.util.ArrayList;
 
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
@@ -70,7 +70,7 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 
 	private final int skipLimit;
 
-	private ExceptionClassifier<String> exceptionClassifier;
+	private ExceptionClassifier<String,Throwable> exceptionClassifier;
 
 	/**
 	 * Convenience constructor that assumes all exception types are skippable
@@ -80,7 +80,7 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 	@SuppressWarnings("unchecked")
 	public LimitCheckingItemSkipPolicy(int skipLimit) {
 		this(skipLimit,
-				new ArrayList<Class<?>>(){{add(Exception.class);}},
+				new HashSet<Class<? extends Throwable>>(){{add(Exception.class);}},
 				Collections.EMPTY_LIST);
 	}
 
@@ -92,14 +92,14 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy {
 	 * (non-critical)
 	 * @param fatalExceptions exception classes that should never be skipped
 	 */
-	public LimitCheckingItemSkipPolicy(int skipLimit, List<Class<?>> skippableExceptions, List<Class<?>>fatalExceptions) {
+	public LimitCheckingItemSkipPolicy(int skipLimit, Collection<Class<? extends Throwable>> skippableExceptions, Collection<Class<? extends Throwable>>fatalExceptions) {
 		this.skipLimit = skipLimit;
 		SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier();
-		Map<Class<?>, String> typeMap = new HashMap<Class<?>, String>();
-		for (Class<?> throwable : skippableExceptions) {
+		Map<Class<? extends Throwable>, String> typeMap = new HashMap<Class<? extends Throwable>, String>();
+		for (Class<? extends Throwable> throwable : skippableExceptions) {
 			typeMap.put(throwable, SKIP);
 		}
-		for (Class<?> throwable : fatalExceptions) {
+		for (Class<? extends Throwable> throwable : fatalExceptions) {
 			typeMap.put(throwable, NEVER_SKIP);
 		}
 		exceptionClassifier.setTypeMap(typeMap);
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java
index 2f68fea82..eedd21333 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java
@@ -40,10 +40,9 @@ public class LimitCheckingItemSkipPolicyTests {
 	
 	@Before
 	public void setUp() throws Exception {
-		List<Class<?>> skippableExceptions = new ArrayList<Class<?>>();
+		List<Class<? extends Throwable>> skippableExceptions = new ArrayList<Class<? extends Throwable>>();
 		skippableExceptions.add(FlatFileParseException.class);
-		List<Class<?>> fatalExceptions = new ArrayList<Class<?>>();
-		
+		List<Class<? extends Throwable>> fatalExceptions = new ArrayList<Class<? extends Throwable>>();
 		failurePolicy = new LimitCheckingItemSkipPolicy(1, skippableExceptions, fatalExceptions);
 	}
 	
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandlerTests.java
index e05e75897..49e7268af 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandlerTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleRetryExceptionHandlerTests.java
@@ -15,9 +15,11 @@
  */
 package org.springframework.batch.core.step.item;
 
+import java.util.Collection;
+import java.util.HashSet;
+
 import junit.framework.TestCase;
 
-import org.springframework.batch.core.step.item.SimpleRetryExceptionHandler;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.context.RepeatContextSupport;
 import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
@@ -37,6 +39,7 @@ public class SimpleRetryExceptionHandlerTests extends TestCase {
 
 	/*
 	 * (non-Javadoc)
+	 * 
 	 * @see junit.framework.TestCase#setUp()
 	 */
 	protected void setUp() throws Exception {
@@ -45,6 +48,7 @@ public class SimpleRetryExceptionHandlerTests extends TestCase {
 
 	/*
 	 * (non-Javadoc)
+	 * 
 	 * @see junit.framework.TestCase#tearDown()
 	 */
 	protected void tearDown() throws Exception {
@@ -53,14 +57,20 @@ public class SimpleRetryExceptionHandlerTests extends TestCase {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.core.step.item.SimpleRetryExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}.
+	 * {@link org.springframework.batch.core.step.item.SimpleRetryExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}
+	 * .
 	 */
 	public void testRethrowWhenRetryExhausted() throws Throwable {
 
 		RetryPolicy retryPolicy = new NeverRetryPolicy();
 		RuntimeException ex = new RuntimeException("foo");
 
-		SimpleRetryExceptionHandler handler = getHandlerAfterRetry(retryPolicy, ex, new Class[] { Error.class });
+		SimpleRetryExceptionHandler handler = getHandlerAfterRetry(retryPolicy, ex,
+				new HashSet<Class<? extends Throwable>>() {
+					{
+						add(Error.class);
+					}
+				});
 
 		// Then pretend to handle the exception in the parent context...
 		try {
@@ -79,14 +89,20 @@ public class SimpleRetryExceptionHandlerTests extends TestCase {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.core.step.item.SimpleRetryExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}.
+	 * {@link org.springframework.batch.core.step.item.SimpleRetryExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}
+	 * .
 	 */
 	public void testNoRethrowWhenRetryNotExhausted() throws Throwable {
 
 		RetryPolicy retryPolicy = new AlwaysRetryPolicy();
 		RuntimeException ex = new RuntimeException("foo");
 
-		SimpleRetryExceptionHandler handler = getHandlerAfterRetry(retryPolicy, ex, new Class[] { Error.class });
+		SimpleRetryExceptionHandler handler = getHandlerAfterRetry(retryPolicy, ex,
+				new HashSet<Class<? extends Throwable>>() {
+					{
+						add(Error.class);
+					}
+				});
 
 		// Then pretend to handle the exception in the parent context...
 		handler.handleException(context.getParent(), ex);
@@ -97,14 +113,20 @@ public class SimpleRetryExceptionHandlerTests extends TestCase {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.core.step.item.SimpleRetryExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}.
+	 * {@link org.springframework.batch.core.step.item.SimpleRetryExceptionHandler#handleException(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)}
+	 * .
 	 */
 	public void testRethrowWhenFatal() throws Throwable {
 
 		RetryPolicy retryPolicy = new AlwaysRetryPolicy();
 		RuntimeException ex = new RuntimeException("foo");
 
-		SimpleRetryExceptionHandler handler = getHandlerAfterRetry(retryPolicy, ex, new Class[] { RuntimeException.class });
+		SimpleRetryExceptionHandler handler = getHandlerAfterRetry(retryPolicy, ex,
+				new HashSet<Class<? extends Throwable>>() {
+					{
+						add(RuntimeException.class);
+					}
+				});
 
 		// Then pretend to handle the exception in the parent context...
 		try {
@@ -125,7 +147,8 @@ public class SimpleRetryExceptionHandlerTests extends TestCase {
 	 * @param ex
 	 * @return
 	 */
-	private SimpleRetryExceptionHandler getHandlerAfterRetry(RetryPolicy retryPolicy, RuntimeException ex, Class<?>[] fatalExceptions) {
+	private SimpleRetryExceptionHandler getHandlerAfterRetry(RetryPolicy retryPolicy, RuntimeException ex,
+			Collection<Class<? extends Throwable>> fatalExceptions) {
 
 		// Always rethrow if the retry is exhausted
 		SimpleRetryExceptionHandler handler = new SimpleRetryExceptionHandler(retryPolicy,
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
index 574054788..01d19181a 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
@@ -9,6 +9,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 
 import org.apache.commons.logging.Log;
@@ -45,7 +46,12 @@ public class SkipLimitStepFactoryBeanTests {
 
 	private SkipLimitStepFactoryBean<String, String> factory = new SkipLimitStepFactoryBean<String, String>();
 
-	private Class<?>[] skippableExceptions = new Class[] { SkippableException.class, SkippableRuntimeException.class };
+	private Collection<Class<? extends Throwable>> skippableExceptions = new HashSet<Class<? extends Throwable>>() {
+		{
+			add(SkippableException.class);
+			add(SkippableRuntimeException.class);
+		}
+	};
 
 	private SkipReaderStub reader = new SkipReaderStub();
 
@@ -133,7 +139,11 @@ public class SkipLimitStepFactoryBeanTests {
 	 */
 	@Test
 	public void testFatalException() throws Exception {
-		factory.setFatalExceptionClasses(new Class[] { FatalRuntimeException.class });
+		factory.setFatalExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(FatalRuntimeException.class);
+			}
+		});
 		factory.setItemWriter(new SkipWriterStub() {
 			public void write(List<? extends String> items) {
 				throw new FatalRuntimeException("Ouch!");
@@ -194,7 +204,11 @@ public class SkipLimitStepFactoryBeanTests {
 
 		factory.setSkipLimit(3);
 		factory.setItemReader(reader);
-		factory.setSkippableExceptionClasses(new Class[] { Exception.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 
 		Step step = (Step) factory.getObject();
 
@@ -238,7 +252,11 @@ public class SkipLimitStepFactoryBeanTests {
 				throw new RuntimeException("oops");
 			}
 		} });
-		factory.setSkippableExceptionClasses(new Class[] { Exception.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 
 		Step step = (Step) factory.getObject();
 
@@ -275,7 +293,11 @@ public class SkipLimitStepFactoryBeanTests {
 				throw new RuntimeException("oops");
 			}
 		} });
-		factory.setSkippableExceptionClasses(new Class[] { Exception.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 
 		Step step = (Step) factory.getObject();
 
@@ -311,15 +333,16 @@ public class SkipLimitStepFactoryBeanTests {
 
 		StepExecution stepExecution = jobExecution.createStepExecution(step);
 
-		// TODO: uncomment this! 
-//		step.execute(stepExecution);
-//		assertEquals(4, stepExecution.getSkipCount());
-//		assertEquals(3, stepExecution.getReadSkipCount());
-//		assertEquals(1, stepExecution.getWriteSkipCount());
-//
-//		// skipped 2,3,4,5
-//		List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6"));
-//		assertEquals(expectedOutput, writer.written);
+		// TODO: uncomment this!
+		// step.execute(stepExecution);
+		// assertEquals(4, stepExecution.getSkipCount());
+		// assertEquals(3, stepExecution.getReadSkipCount());
+		// assertEquals(1, stepExecution.getWriteSkipCount());
+		//
+		// // skipped 2,3,4,5
+		// List<String> expectedOutput =
+		// Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6"));
+		// assertEquals(expectedOutput, writer.written);
 
 	}
 
@@ -343,21 +366,26 @@ public class SkipLimitStepFactoryBeanTests {
 
 		StepExecution stepExecution = jobExecution.createStepExecution(step);
 
-		// TODO: uncomment this! 
-//		step.execute(stepExecution);
-//		assertEquals(4, stepExecution.getSkipCount());
-//		assertEquals(2, stepExecution.getReadSkipCount());
-//		assertEquals(2, stepExecution.getWriteSkipCount());
-//
-//		// skipped 2,3,4,5
-//		List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7"));
-//		assertEquals(expectedOutput, writer.written);
+		// TODO: uncomment this!
+		// step.execute(stepExecution);
+		// assertEquals(4, stepExecution.getSkipCount());
+		// assertEquals(2, stepExecution.getReadSkipCount());
+		// assertEquals(2, stepExecution.getWriteSkipCount());
+		//
+		// // skipped 2,3,4,5
+		// List<String> expectedOutput =
+		// Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7"));
+		// assertEquals(expectedOutput, writer.written);
 
 	}
 
 	@Test
 	public void testDefaultSkipPolicy() throws Exception {
-		factory.setSkippableExceptionClasses(new Class[] { Exception.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 		factory.setSkipLimit(1);
 		List<String> items = Arrays.asList(new String[] { "a", "b", "c" });
 		ItemReader<String> provider = new ListItemReader<String>(items) {
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
index 0142e36ef..4f623bc0d 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.fail;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Date;
+import java.util.HashSet;
 import java.util.List;
 
 import org.apache.commons.logging.Log;
@@ -100,7 +101,11 @@ public class StatefulRetryStepFactoryBeanTests {
 		factory.setItemWriter(processor);
 		factory.setJobRepository(repository);
 		factory.setTransactionManager(new ResourcelessTransactionManager());
-		factory.setRetryableExceptionClasses(new Class[] { Exception.class });
+		factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 		factory.setCommitInterval(1); // trivial by default
 
 		JobSupport job = new JobSupport("jobName");
@@ -146,7 +151,7 @@ public class StatefulRetryStepFactoryBeanTests {
 		};
 		factory.setItemReader(provider);
 		factory.setRetryLimit(10);
-		factory.setSkippableExceptionClasses(new Class[0]);
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>());
 		Step step = (Step) factory.getObject();
 
 		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -166,7 +171,11 @@ public class StatefulRetryStepFactoryBeanTests {
 
 	@Test
 	public void testSkipAndRetry() throws Exception {
-		factory.setSkippableExceptionClasses(new Class[] { Exception.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 		factory.setSkipLimit(2);
 		List<String> items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" });
 		ItemReader<String> provider = new ListItemReader<String>(items) {
@@ -195,7 +204,11 @@ public class StatefulRetryStepFactoryBeanTests {
 	@Test
 	public void testSkipAndRetryWithWriteFailure() throws Exception {
 
-		factory.setSkippableExceptionClasses(new Class[] { RetryException.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(RetryException.class);
+			}
+		});
 		factory.setListeners(new StepListener[] { new SkipListenerSupport() {
 			public void onSkipInWrite(Object item, Throwable t) {
 				recovered.add(item);
@@ -226,7 +239,11 @@ public class StatefulRetryStepFactoryBeanTests {
 		factory.setItemReader(provider);
 		factory.setItemWriter(itemWriter);
 		factory.setRetryLimit(5);
-		factory.setRetryableExceptionClasses(new Class[] { RuntimeException.class });
+		factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(RuntimeException.class);
+			}
+		});
 		AbstractStep step = (AbstractStep) factory.getObject();
 		step.setName("mytest");
 		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -248,7 +265,11 @@ public class StatefulRetryStepFactoryBeanTests {
 	public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval() throws Exception {
 
 		factory.setCommitInterval(3);
-		factory.setSkippableExceptionClasses(new Class[] { RetryException.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(RetryException.class);
+			}
+		});
 		factory.setListeners(new StepListener[] { new SkipListenerSupport() {
 			public void onSkipInWrite(Object item, Throwable t) {
 				recovered.add(item);
@@ -279,7 +300,11 @@ public class StatefulRetryStepFactoryBeanTests {
 		factory.setItemReader(provider);
 		factory.setItemWriter(itemWriter);
 		factory.setRetryLimit(5);
-		factory.setRetryableExceptionClasses(new Class[] { RuntimeException.class });
+		factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(RuntimeException.class);
+			}
+		});
 		AbstractStep step = (AbstractStep) factory.getObject();
 		step.setName("mytest");
 		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -300,7 +325,11 @@ public class StatefulRetryStepFactoryBeanTests {
 
 	@Test
 	public void testRetryWithNoSkip() throws Exception {
-		factory.setRetryableExceptionClasses(new Class[] { Exception.class });
+		factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(Exception.class);
+			}
+		});
 		factory.setRetryLimit(4);
 		factory.setSkipLimit(0);
 		List<String> items = Arrays.asList(new String[] { "b" });
@@ -346,9 +375,13 @@ public class StatefulRetryStepFactoryBeanTests {
 	public void testNonSkippableException() throws Exception {
 
 		// Very specific skippable exception
-		factory.setSkippableExceptionClasses(new Class[] { UnsupportedOperationException.class });
+		factory.setSkippableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(UnsupportedOperationException.class);
+			}
+		});
 		// ...which is not retryable...
-		factory.setRetryableExceptionClasses(new Class<?>[0]);
+		factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>());
 
 		factory.setSkipLimit(1);
 		List<String> items = Arrays.asList(new String[] { "b" });
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/LogOrRethrowExceptionHandler.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/LogOrRethrowExceptionHandler.java
index 5c792b753..7623d9ab5 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/LogOrRethrowExceptionHandler.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/LogOrRethrowExceptionHandler.java
@@ -56,7 +56,7 @@ public class LogOrRethrowExceptionHandler implements ExceptionHandler {
 
 	protected final Log logger = LogFactory.getLog(LogOrRethrowExceptionHandler.class);
 
-	private ExceptionClassifier<String> exceptionClassifier = new ExceptionClassifierSupport() {
+	private ExceptionClassifier<String,Throwable> exceptionClassifier = new ExceptionClassifierSupport() {
 		public String classify(Throwable throwable) {
 			return RETHROW;
 		}
@@ -68,7 +68,7 @@ public class LogOrRethrowExceptionHandler implements ExceptionHandler {
 	 * 
 	 * @param exceptionClassifier the ExceptionClassifier to use
 	 */
-	public void setExceptionClassifier(ExceptionClassifier<String> exceptionClassifier) {
+	public void setExceptionClassifier(ExceptionClassifier<String,Throwable> exceptionClassifier) {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 
@@ -81,7 +81,7 @@ public class LogOrRethrowExceptionHandler implements ExceptionHandler {
 	 */
 	public void handleException(RepeatContext context, Throwable throwable) throws Throwable {
 
-		Object key = exceptionClassifier.classify(throwable);
+		String key = exceptionClassifier.classify(throwable);
 		if (ERROR.equals(key)) {
 			logger.error("Exception encountered in batch repeat.", throwable);
 		} else if (WARN.equals(key)) {
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandler.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandler.java
index d0d361dd0..e7f8a3f8d 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandler.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandler.java
@@ -18,7 +18,6 @@ package org.springframework.batch.repeat.exception;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Map.Entry;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -26,7 +25,6 @@ import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.context.RepeatContextCounter;
 import org.springframework.batch.support.ExceptionClassifier;
 import org.springframework.batch.support.ExceptionClassifierSupport;
-import org.springframework.util.Assert;
 
 /**
  * Implementation of {@link ExceptionHandler} that rethrows when exceptions of a
@@ -41,9 +39,9 @@ public class RethrowOnThresholdExceptionHandler implements ExceptionHandler {
 
 	protected final Log logger = LogFactory.getLog(RethrowOnThresholdExceptionHandler.class);
 
-	private ExceptionClassifier<String> exceptionClassifier = new ExceptionClassifierSupport();
+	private ExceptionClassifier<String,Throwable> exceptionClassifier = new ExceptionClassifierSupport();
 
-	private Map<Object, Integer> thresholds = new HashMap<Object, Integer>();
+	private Map<String, Integer> thresholds = new HashMap<String, Integer>();
 
 	private boolean useParent = false;
 
@@ -75,15 +73,7 @@ public class RethrowOnThresholdExceptionHandler implements ExceptionHandler {
 	 * 
 	 * @param thresholds the threshold value map.
 	 */
-	public void setThresholds(Map<Object, Integer> thresholds) {
-		for (Entry<Object, Integer> entry : thresholds.entrySet()) {
-
-			if (!(entry.getKey() instanceof String)) {
-				logger.warn("Key in thresholds map is not of type String: " + entry.getKey());
-			}
-			Assert.state(entry.getValue() != null, "Threshold value must be of type Integer.  "
-					+ "Try using the value-type attribute if you care configuring this map via xml.");
-		}
+	public void setThresholds(Map<String, Integer> thresholds) {
 		this.thresholds = thresholds;
 	}
 
@@ -95,7 +85,7 @@ public class RethrowOnThresholdExceptionHandler implements ExceptionHandler {
 	 * 
 	 * @param exceptionClassifier ExceptionClassifier to use
 	 */
-	public void setExceptionClassifier(ExceptionClassifier<String> exceptionClassifier) {
+	public void setExceptionClassifier(ExceptionClassifier<String,Throwable> exceptionClassifier) {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/SimpleLimitExceptionHandler.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/SimpleLimitExceptionHandler.java
index 03add5aa8..768d1a223 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/SimpleLimitExceptionHandler.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/exception/SimpleLimitExceptionHandler.java
@@ -118,7 +118,7 @@ public class SimpleLimitExceptionHandler implements ExceptionHandler {
 	 * @param limit the limit
 	 */
 	public void setLimit(final int limit) {
-		delegate.setThresholds(new HashMap<Object, Integer>() {
+		delegate.setThresholds(new HashMap<String, Integer>() {
 			{
 				put(ExceptionClassifierSupport.DEFAULT, 0);
 				put(TX_INVALID, limit);
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java
index 4c7a3feb6..968292a6d 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ExceptionClassifierRetryPolicy.java
@@ -37,7 +37,7 @@ import org.springframework.util.Assert;
  */
 public class ExceptionClassifierRetryPolicy extends AbstractStatelessRetryPolicy {
 
-	private ExceptionClassifier<String> exceptionClassifier = new ExceptionClassifierSupport();
+	private ExceptionClassifier<String,Throwable> exceptionClassifier = new ExceptionClassifierSupport();
 
 	private Map<String, RetryPolicy> policyMap = new HashMap<String, RetryPolicy>();
 
@@ -64,7 +64,7 @@ public class ExceptionClassifierRetryPolicy extends AbstractStatelessRetryPolicy
 	 * 
 	 * @param exceptionClassifier ExceptionClassifier to use
 	 */
-	public void setExceptionClassifier(ExceptionClassifier<String> exceptionClassifier) {
+	public void setExceptionClassifier(ExceptionClassifier<String,Throwable> exceptionClassifier) {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 
@@ -113,7 +113,7 @@ public class ExceptionClassifierRetryPolicy extends AbstractStatelessRetryPolicy
 
 	private class ExceptionClassifierRetryContext extends RetryContextSupport implements RetryPolicy {
 
-		private ExceptionClassifier<String> exceptionClassifier;
+		private ExceptionClassifier<String,Throwable> exceptionClassifier;
 
 		// Dynamic: depends on the latest exception:
 		RetryPolicy policy;
@@ -126,7 +126,7 @@ public class ExceptionClassifierRetryPolicy extends AbstractStatelessRetryPolicy
 
 		Map<RetryPolicy, RetryContext> contexts = new HashMap<RetryPolicy, RetryContext>();
 
-		public ExceptionClassifierRetryContext(RetryContext parent, ExceptionClassifier<String> exceptionClassifier) {
+		public ExceptionClassifierRetryContext(RetryContext parent, ExceptionClassifier<String,Throwable> exceptionClassifier) {
 			super(parent);
 			this.exceptionClassifier = exceptionClassifier;
 			Object key = exceptionClassifier.getDefault();
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
index 22383f74d..885bfdd63 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
@@ -16,6 +16,9 @@
 
 package org.springframework.batch.retry.policy;
 
+import java.util.Collection;
+import java.util.HashSet;
+
 import org.springframework.batch.retry.RetryCallback;
 import org.springframework.batch.retry.RetryContext;
 import org.springframework.batch.retry.context.RetryContextSupport;
@@ -67,8 +70,13 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 	 */
 	public SimpleRetryPolicy(int maxAttempts) {
 		super();
-		setRetryableExceptionClasses(new Class[] { Exception.class });
-		setFatalExceptionClasses(new Class[] { Error.class });
+		Collection<Class<? extends Throwable>> classes;
+		classes = new HashSet<Class<? extends Throwable>>();
+		classes.add(Exception.class);
+		setRetryableExceptionClasses(classes);
+		classes = new HashSet<Class<? extends Throwable>>();
+		classes.add(Error.class);
+		setFatalExceptionClasses(classes);
 		this.maxAttempts = maxAttempts;
 	}
 
@@ -103,7 +111,7 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 	 * 
 	 * @param retryableExceptionClasses defaults to {@link Exception}.
 	 */
-	public final void setRetryableExceptionClasses(Class<?>[] retryableExceptionClasses) {
+	public final void setRetryableExceptionClasses(Collection<Class<? extends Throwable>> retryableExceptionClasses) {
 		retryableClassifier.setExceptionClasses(retryableExceptionClasses);
 	}
 
@@ -114,7 +122,7 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
 	 * 
 	 * @param fatalExceptionClasses defaults to {@link Exception}.
 	 */
-	public final void setFatalExceptionClasses(Class<?>[] fatalExceptionClasses) {
+	public final void setFatalExceptionClasses(Collection<Class<? extends Throwable>> fatalExceptionClasses) {
 		fatalClassifier.setExceptionClasses(fatalExceptionClasses);
 	}
 
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/BinaryExceptionClassifier.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/BinaryExceptionClassifier.java
index 0e6440f8a..db136bcd1 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/BinaryExceptionClassifier.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/BinaryExceptionClassifier.java
@@ -15,6 +15,7 @@
  */
 package org.springframework.batch.support;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -41,9 +42,9 @@ public class BinaryExceptionClassifier extends ExceptionClassifierSupport {
 	 * 
 	 * @param exceptionClasses defaults to {@link Exception}.
 	 */
-	public final void setExceptionClasses(Class<?>[] exceptionClasses) {
-		Map<Class<?>, String> temp = new HashMap<Class<?>, String>();
-		for (Class<?> exceptionClass : exceptionClasses) {
+	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);
 		}
 		this.delegate.setTypeMap(temp);
@@ -55,7 +56,7 @@ public class BinaryExceptionClassifier extends ExceptionClassifierSupport {
 	 * 
 	 * @param throwable the Throwable to classify
 	 * @return true if it is default classified (i.e. not on the list provided
-	 * in {@link #setExceptionClasses(Class[])}.
+	 * in {@link #setExceptionClasses(Collection)}.
 	 */
 	public boolean isDefault(Throwable throwable) {
 		return classify(throwable).equals(DEFAULT);
@@ -67,7 +68,7 @@ public class BinaryExceptionClassifier extends ExceptionClassifierSupport {
 	 * of the throwable or one of its ancestors is on the exception class list
 	 * the classification is as {@link #NON_DEFAULT}.
 	 * 
-	 * @see #setExceptionClasses(Class[])
+	 * @see #setExceptionClasses(Collection)
 	 * @see ExceptionClassifierSupport#classify(Throwable)
 	 */
 	public String classify(Throwable throwable) {
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifier.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifier.java
index aaf3416fa..4de99695b 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifier.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifier.java
@@ -22,24 +22,24 @@ package org.springframework.batch.support;
  * @author Dave Syer
  * 
  */
-public interface ExceptionClassifier<T> {
+public interface ExceptionClassifier<T,C> {
 
 	/**
 	 * Get a default value, normally the same as would be returned by
-	 * {@link #classify(Throwable)} with null argument.
+	 * {@link #classify(Object)} with null argument.
 	 * 
 	 * @return the default value.
 	 */
 	T getDefault();
 
 	/**
-	 * Classify the given exception and return a non-null object. The return
+	 * 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.
 	 * 
-	 * @param throwable the input exception. Can be null.
+	 * @param classifiable the input object. Can be null.
 	 * @return an object.
 	 */
-	T classify(Throwable throwable);
+	T classify(C classifiable);
 
 }
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifierSupport.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifierSupport.java
index 6f291c3b3..e0b5f6864 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifierSupport.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/ExceptionClassifierSupport.java
@@ -23,7 +23,7 @@ package org.springframework.batch.support;
  * @author Dave Syer
  * 
  */
-public class ExceptionClassifierSupport implements ExceptionClassifier<String> {
+public class ExceptionClassifierSupport implements ExceptionClassifier<String,Throwable> {
 
 	/**
 	 * Default classification key.
@@ -33,7 +33,7 @@ public class ExceptionClassifierSupport implements ExceptionClassifier<String> {
 	/**
 	 * Always returns the value of {@link #DEFAULT}.
 	 * 
-	 * @see org.springframework.batch.support.ExceptionClassifier#classify(java.lang.Throwable)
+	 * @see org.springframework.batch.support.ExceptionClassifier#classify(Object)
 	 */
 	public String classify(Throwable throwable) {
 		return DEFAULT;
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SubclassExceptionClassifier.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SubclassExceptionClassifier.java
index f6ad86ce8..249d7b875 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SubclassExceptionClassifier.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SubclassExceptionClassifier.java
@@ -30,7 +30,7 @@ import org.springframework.util.Assert;
  */
 public class SubclassExceptionClassifier extends ExceptionClassifierSupport {
 	
-	private Map<Class<?>, String> classified = new HashMap<Class<?>, String>();
+	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
@@ -39,9 +39,9 @@ public class SubclassExceptionClassifier extends ExceptionClassifierSupport {
 	 * 
 	 * @param typeMap the typeMap to set
 	 */
-	public final void setTypeMap(Map<Class<?>, String> typeMap) {
-		Map<Class<?>, String> map = new HashMap<Class<?>, String>();
-		for (Map.Entry<Class<?>, String> entry : typeMap.entrySet()) {
+	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;
@@ -59,15 +59,15 @@ public class SubclassExceptionClassifier extends ExceptionClassifierSupport {
 			return super.classify(throwable);
 		}
 
-		Class<?> exceptionClass = throwable.getClass();
+		Class<? extends Throwable> exceptionClass = throwable.getClass();
 		if (classified.containsKey(exceptionClass)) {
 			return classified.get(exceptionClass);
 		}
 
 		// check for subclasses
-		Set<Class<?>> classes = new TreeSet<Class<?>>(new ClassComparator());
+		Set<Class<? extends Throwable>> classes = new TreeSet<Class<? extends Throwable>>(new ClassComparator());
 		classes.addAll(classified.keySet());
-		for (Class<?> cls : classes) {
+		for (Class<? extends Throwable> cls : classes) {
 			if (cls.isAssignableFrom(exceptionClass)) {
 				String value = classified.get(cls);
 				addRetryableExceptionClass(exceptionClass, value, this.classified);
@@ -78,7 +78,7 @@ public class SubclassExceptionClassifier extends ExceptionClassifierSupport {
 		return super.classify(throwable);
 	}
 
-	private void addRetryableExceptionClass(Class<?> exceptionClass, String classifiedAs, Map<Class<?>, String> map) {
+	private void addRetryableExceptionClass(Class<? extends Throwable> exceptionClass, String classifiedAs, Map<Class<? extends Throwable>, String> map) {
 		Assert.isAssignable(Throwable.class, exceptionClass);
 		map.put(exceptionClass, classifiedAs);
 	}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandlerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandlerTests.java
index 4ccbd632e..093b62a8f 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandlerTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/exception/RethrowOnThresholdExceptionHandlerTests.java
@@ -55,7 +55,7 @@ public class RethrowOnThresholdExceptionHandlerTests extends TestCase {
 				return "RuntimeException";
 			}
 		});
-		handler.setThresholds(Collections.singletonMap((Object)"RuntimeException", new Integer(1)));
+		handler.setThresholds(Collections.singletonMap("RuntimeException", new Integer(1)));
 		// No exception...
 		handler.handleException(context, new RuntimeException("Foo"));
 		RepeatContextCounter counter = new RepeatContextCounter(context, RethrowOnThresholdExceptionHandler.class.getName() + ".RuntimeException");
@@ -69,7 +69,7 @@ public class RethrowOnThresholdExceptionHandlerTests extends TestCase {
 				return "RuntimeException";
 			}
 		});
-		handler.setThresholds(Collections.singletonMap((Object)"RuntimeException", new Integer(2)));
+		handler.setThresholds(Collections.singletonMap("RuntimeException", new Integer(2)));
 		// No exception...
 		handler.handleException(context, new RuntimeException("Foo"));
 		handler.handleException(context, new RuntimeException("Foo"));
@@ -88,7 +88,7 @@ public class RethrowOnThresholdExceptionHandlerTests extends TestCase {
 				return "RuntimeException";
 			}
 		});
-		handler.setThresholds(Collections.singletonMap((Object)"RuntimeException", new Integer(1)));
+		handler.setThresholds(Collections.singletonMap("RuntimeException", new Integer(1)));
 		// No exception...
 		handler.handleException(context, new RuntimeException("Foo"));
 		context = new RepeatContextSupport(parent);
@@ -107,7 +107,7 @@ public class RethrowOnThresholdExceptionHandlerTests extends TestCase {
 				return "RuntimeException";
 			}
 		});
-		handler.setThresholds(Collections.singletonMap((Object)"RuntimeException", new Integer(1)));
+		handler.setThresholds(Collections.singletonMap("RuntimeException", new Integer(1)));
 		handler.setUseParent(true);
 		// No exception...
 		handler.handleException(context, new RuntimeException("Foo"));
@@ -120,16 +120,5 @@ public class RethrowOnThresholdExceptionHandlerTests extends TestCase {
 			assertEquals("Foo", e.getMessage());
 		}
 	}
-	
-	public void testNotStringAsKey() throws Exception {
-		try {
-			handler.setThresholds(Collections.singletonMap((Object)RuntimeException.class, new Integer(1)));
-			// It's not an error, but not advised...
-		}
-		catch (RuntimeException e) {
-			throw e;
-		}
-		
-	}
 
 }
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java
index 264647542..9a0c18ff4 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java
@@ -16,35 +16,30 @@
 
 package org.springframework.batch.retry.policy;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+
 import junit.framework.TestCase;
 
 import org.springframework.batch.retry.RetryContext;
 
 public class SimpleRetryPolicyTests extends TestCase {
 
-	public void testSetInvalidExceptionClass() throws Exception {
-		try {
-			new SimpleRetryPolicy().setRetryableExceptionClasses(new Class[] { String.class });
-			fail("Should only be able to set Exception classes.");
-		}
-		catch (IllegalArgumentException ex) {
-
-		}
-	}
-
 	public void testCanRetryIfNoException() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null, null);
 		assertTrue(policy.canRetry(context));
 	}
 
+	@SuppressWarnings("unchecked")
 	public void testEmptyExceptionsNeverRetry() throws Exception {
 
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null, null);
 
 		// We can't retry any exceptions...
-		policy.setRetryableExceptionClasses(new Class[0]);
+		policy.setRetryableExceptionClasses(Collections.EMPTY_SET);
 
 		// ...so we can't retry this one...
 		policy.registerThrowable(context, new IllegalStateException());
@@ -92,14 +87,24 @@ public class SimpleRetryPolicyTests extends TestCase {
 
 	public void testFatalOverridesRetryable() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
-		policy.setFatalExceptionClasses(new Class[] {Exception.class});
-		policy.setRetryableExceptionClasses(new Class[] {RuntimeException.class});
+		policy.setFatalExceptionClasses(getClasses(Exception.class));
+		policy.setRetryableExceptionClasses(getClasses(RuntimeException.class));
 		RetryContext context = policy.open(null, null);
 		assertNotNull(context);
 		policy.registerThrowable(context, new RuntimeException("foo"));
 		assertFalse(policy.canRetry(context));
 	}
 
+	/**
+	 * @param cls
+	 * @return
+	 */
+	private Collection<Class<? extends Throwable>> getClasses(Class<? extends Throwable> cls) {
+		Collection<Class<? extends Throwable>> classes = new HashSet<Class<? extends Throwable>>();
+		classes.add(cls);
+		return classes;
+	}
+
 	public void testParent() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null, null);
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/support/RetryTemplateTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/support/RetryTemplateTests.java
index 693d2d4f3..039507fc6 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/support/RetryTemplateTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/support/RetryTemplateTests.java
@@ -16,6 +16,8 @@
 
 package org.springframework.batch.retry.support;
 
+import java.util.HashSet;
+
 import junit.framework.TestCase;
 
 import org.springframework.batch.retry.ExhaustedRetryException;
@@ -96,7 +98,11 @@ public class RetryTemplateTests extends TestCase {
 		RetryTemplate template = new RetryTemplate();
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		template.setRetryPolicy(policy);
-		policy.setRetryableExceptionClasses(new Class[] { RuntimeException.class });
+		policy.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(RuntimeException.class);
+			}
+		});
 
 		int attempts = 3;
 
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/BinaryExceptionClassifierTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/BinaryExceptionClassifierTests.java
index 13c37cfa7..de5611288 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/BinaryExceptionClassifierTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/BinaryExceptionClassifierTests.java
@@ -16,7 +16,7 @@
 
 package org.springframework.batch.support;
 
-import org.springframework.batch.support.BinaryExceptionClassifier;
+import java.util.HashSet;
 
 import junit.framework.TestCase;
 
@@ -33,8 +33,11 @@ public class BinaryExceptionClassifierTests extends TestCase {
 	}
 
 	public void testClassifyExactMatch() {
-		classifier.setExceptionClasses(new Class[] {IllegalStateException.class});
+		classifier.setExceptionClasses(new HashSet<Class<? extends Throwable>>() {
+			{
+				add(IllegalStateException.class);
+			}
+		});
 		assertEquals(false, classifier.isDefault(new IllegalStateException("Foo")));
 	}
-
 }
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/SubclassExceptionClassifierTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/SubclassExceptionClassifierTests.java
index 9b7798050..9d4087cf2 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/SubclassExceptionClassifierTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/SubclassExceptionClassifierTests.java
@@ -33,28 +33,28 @@ public class SubclassExceptionClassifierTests extends TestCase {
 	}
 
 	public void testClassifyExactMatch() {
-		classifier.setTypeMap(new LinkedHashMap<Class<?>, String>() {{
+		classifier.setTypeMap(new LinkedHashMap<Class<? extends Throwable>, String>() {{
 			put(IllegalStateException.class, "foo");
 		}});
 		assertEquals("foo", classifier.classify(new IllegalStateException("Foo")));
 	}
 
 	public void testClassifySubclassMatch() {
-		classifier.setTypeMap(new LinkedHashMap<Class<?>, String>() {{
+		classifier.setTypeMap(new LinkedHashMap<Class<? extends Throwable>, String>() {{
 			put(RuntimeException.class, "foo");
 		}});
 		assertEquals("foo", classifier.classify(new IllegalStateException("Foo")));
 	}
 
 	public void testClassifySuperclassDoesNotMatch() {
-		classifier.setTypeMap(new LinkedHashMap<Class<?>, String>() {{
+		classifier.setTypeMap(new LinkedHashMap<Class<? extends Throwable>, String>() {{
 			put(IllegalStateException.class, "foo");
 		}});
 		assertEquals(classifier.getDefault(), classifier.classify(new RuntimeException("Foo")));
 	}
 
 	public void testClassifyAncestorMatch() {
-		classifier.setTypeMap(new LinkedHashMap<Class<?>, String>() {{
+		classifier.setTypeMap(new LinkedHashMap<Class<? extends Throwable>, String>() {{
 			put(Exception.class, "bar");
 			put(IllegalArgumentException.class, "foo");
 			put(RuntimeException.class, "bucket");