diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java
index 9775fe436..fa9491230 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java
@@ -73,9 +73,9 @@ public class RetryTemplate implements RetryOperations {
 	private volatile RetryListener[] listeners = new RetryListener[0];
 
 	/**
-	 * Setter for listeners. The listeners are executed before and after a
-	 * retry block (i.e. before and after all the attempts), and on an error
-	 * (every attempt).
+	 * Setter for listeners. The listeners are executed before and after a retry
+	 * block (i.e. before and after all the attempts), and on an error (every
+	 * attempt).
 	 * @param listeners
 	 * @see RetryListener
 	 */
@@ -168,17 +168,17 @@ public class RetryTemplate implements RetryOperations {
 					lastException = null;
 					return callback.doWithRetry(context);
 				}
-				catch (Throwable e) {
+				catch (Throwable ex) {
+					Throwable throwable = unwrapIfRethrown(ex);
+					lastException = throwable;
 
-					lastException = e;
+					doOnErrorInterceptors(callback, context, throwable);
 
-					doOnErrorInterceptors(callback, context, e);
-
-					retryPolicy.registerThrowable(context, e);
+					retryPolicy.registerThrowable(context, throwable);
 
 					if (retryPolicy.shouldRethrow(context)) {
 						logger.debug("Abort retry for policy: count=" + context.getRetryCount());
-						rethrow(e);
+						rethrow(throwable);
 					}
 
 				}
@@ -245,16 +245,44 @@ public class RetryTemplate implements RetryOperations {
 		}
 	}
 
-	private static void rethrow(Throwable ex) throws Exception {
-		if (ex instanceof Exception) {
-			throw (Exception) ex;
+	/**
+	 * Re-throw the exception directly if possible, wrap custom Throwables into
+	 * {@link UnclassifiedRetryException}.
+	 */
+	private static void rethrow(Throwable throwable) throws Exception {
+		if (throwable instanceof Exception) {
+			throw (Exception) throwable;
 		}
-		else if (ex instanceof Error) {
-			throw (Error) ex;
+		else if (throwable instanceof Error) {
+			throw (Error) throwable;
 		}
 		else {
-			throw new RetryException("Unclassified Throwable encountered", ex);
+			throw new UnclassifiedRetryException("Unclassified Throwable encountered", throwable);
 		}
 	}
 
+	/**
+	 * Undo the wrapping done in {@link #rethrow(Throwable)}
+	 */
+	private static Throwable unwrapIfRethrown(Throwable throwable) {
+		if (throwable instanceof UnclassifiedRetryException) {
+			return throwable.getCause();
+		}
+		else {
+			return throwable;
+		}
+	}
+
+	/**
+	 * Runtime exception wrapper for Throwables that are neither Exception nor
+	 * Error.
+	 */
+	private static class UnclassifiedRetryException extends RetryException {
+
+		public UnclassifiedRetryException(String msg, Throwable cause) {
+			super(msg, cause);
+		}
+
+	}
+
 }
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 8c395976b..693d2d4f3 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
@@ -22,10 +22,12 @@ import org.springframework.batch.retry.ExhaustedRetryException;
 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.backoff.BackOffContext;
 import org.springframework.batch.retry.backoff.BackOffInterruptedException;
 import org.springframework.batch.retry.backoff.BackOffPolicy;
 import org.springframework.batch.retry.backoff.StatelessBackOffPolicy;
+import org.springframework.batch.retry.listener.RetryListenerSupport;
 import org.springframework.batch.retry.policy.NeverRetryPolicy;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
 
@@ -242,16 +244,61 @@ public class RetryTemplateTests extends TestCase {
 			}
 		};
 		RetryTemplate template = new RetryTemplate();
-		
+
 		try {
 			template.execute(callback);
 			fail();
 		}
 		catch (RetryException expected) {
-			expected.getMessage().equals("Unclassified Throwable encountered");
-			expected.getCause().getMessage().equals("throwable in callback");
+			assertTrue(expected.getMessage().contains("Unclassified Throwable encountered"));
+			assertEquals("throwable in callback", expected.getCause().getMessage());
 		}
-		
+	}
+
+	/**
+	 * If nested template wraps unclassified Throwable into RetryException the
+	 * Throwable is unwrapped before passed to collaborators.
+	 */
+	public void testThrowableUnwrapping() throws Exception {
+
+		final RetryCallback throwingCallback = new RetryCallback() {
+			public Object doWithRetry(RetryContext context) throws Throwable {
+				throw new Throwable("Crashed terribly");
+			}
+		};
+		final RetryTemplate nested = new RetryTemplate();
+
+		RetryCallback callNested = new RetryCallback() {
+			public Object doWithRetry(RetryContext context) throws Throwable {
+				return nested.execute(throwingCallback);
+			}
+		};
+		ExceptionCheckingListener listener = new ExceptionCheckingListener();
+		RetryTemplate template = new RetryTemplate();
+		template.setListeners(new RetryListener[] { listener });
+
+		try {
+			template.execute(callNested);
+			fail();
+		}
+		catch (RetryException expected) {
+			assertTrue(expected.getMessage().contains("Unclassified Throwable encountered"));
+			assertEquals("Crashed terribly", expected.getCause().getMessage());
+		}
+		assertTrue(listener.called);
+	}
+
+	private static class ExceptionCheckingListener extends RetryListenerSupport {
+
+		boolean called = false;
+
+		public void onError(RetryContext context, RetryCallback callback, Throwable throwable) {
+			called = true;
+			assertFalse(throwable instanceof Exception);
+			assertFalse(throwable instanceof Error);
+			assertEquals("Crashed terribly", throwable.getMessage());
+		}
+
 	}
 
 	private static class MockRetryCallback implements RetryCallback {