diff --git a/spring-aop/src/main/java/org/springframework/aop/interceptor/AsyncExecutionInterceptor.java b/spring-aop/src/main/java/org/springframework/aop/interceptor/AsyncExecutionInterceptor.java
index 1a89a40977..8b98bcce66 100644
--- a/spring-aop/src/main/java/org/springframework/aop/interceptor/AsyncExecutionInterceptor.java
+++ b/spring-aop/src/main/java/org/springframework/aop/interceptor/AsyncExecutionInterceptor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.aop.interceptor;
 
 import java.lang.reflect.Method;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
 
@@ -112,6 +113,9 @@ public class AsyncExecutionInterceptor extends AsyncExecutionAspectSupport
 						return ((Future<?>) result).get();
 					}
 				}
+				catch (ExecutionException ex) {
+					handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
+				}
 				catch (Throwable ex) {
 					handleError(ex, userDeclaredMethod, invocation.getArguments());
 				}
diff --git a/spring-context/src/main/java/org/springframework/scheduling/annotation/AsyncResult.java b/spring-context/src/main/java/org/springframework/scheduling/annotation/AsyncResult.java
index 445fcb607f..ae6ecc4271 100644
--- a/spring-context/src/main/java/org/springframework/scheduling/annotation/AsyncResult.java
+++ b/spring-context/src/main/java/org/springframework/scheduling/annotation/AsyncResult.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 
 package org.springframework.scheduling.annotation;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.util.concurrent.FailureCallback;
@@ -31,21 +33,37 @@ import org.springframework.util.concurrent.SuccessCallback;
  * plain {@link java.util.concurrent.Future}, along with the corresponding support
  * in {@code @Async} processing.
  *
+ * <p>As of Spring 4.2, this class also supports passing execution exceptions back
+ * to the caller.
+ *
  * @author Juergen Hoeller
  * @since 3.0
  * @see Async
+ * @see #forValue(Object)
+ * @see #forExecutionException(Throwable)
  */
 public class AsyncResult<V> implements ListenableFuture<V> {
 
 	private final V value;
 
+	private final ExecutionException executionException;
+
 
 	/**
 	 * Create a new AsyncResult holder.
 	 * @param value the value to pass through
 	 */
 	public AsyncResult(V value) {
+		this(value, null);
+	}
+
+	/**
+	 * Create a new AsyncResult holder.
+	 * @param value the value to pass through
+	 */
+	private AsyncResult(V value, ExecutionException ex) {
 		this.value = value;
+		this.executionException = ex;
 	}
 
 
@@ -65,13 +83,16 @@ public class AsyncResult<V> implements ListenableFuture<V> {
 	}
 
 	@Override
-	public V get() {
+	public V get() throws ExecutionException {
+		if (this.executionException != null) {
+			throw this.executionException;
+		}
 		return this.value;
 	}
 
 	@Override
-	public V get(long timeout, TimeUnit unit) {
-		return this.value;
+	public V get(long timeout, TimeUnit unit) throws ExecutionException {
+		return get();
 	}
 
 	@Override
@@ -89,4 +110,28 @@ public class AsyncResult<V> implements ListenableFuture<V> {
 		}
 	}
 
+
+	/**
+	 * Create a new async result which exposes the given value from {@link Future#get()}.
+	 * @param value the value to expose
+	 * @since 4.2
+	 * @see Future#get()
+	 */
+	public static <V> ListenableFuture<V> forValue(V value) {
+		return new AsyncResult<V>(value, null);
+	}
+
+	/**
+	 * Create a new async result which exposes the given exception as an
+	 * {@link ExecutionException} from {@link Future#get()}.
+	 * @param ex the exception to expose (either an pre-built {@link ExecutionException}
+	 * or a cause to be wrapped in an {@link ExecutionException})
+	 * @since 4.2
+	 * @see ExecutionException
+	 */
+	public static <V> ListenableFuture<V> forExecutionException(Throwable ex) {
+		return new AsyncResult<V>(null,
+				(ex instanceof ExecutionException ? (ExecutionException) ex : new ExecutionException(ex)));
+	}
+
 }
diff --git a/spring-context/src/test/java/org/springframework/scheduling/annotation/AsyncExecutionTests.java b/spring-context/src/test/java/org/springframework/scheduling/annotation/AsyncExecutionTests.java
index 2e503211ef..9562fed9e4 100644
--- a/spring-context/src/test/java/org/springframework/scheduling/annotation/AsyncExecutionTests.java
+++ b/spring-context/src/test/java/org/springframework/scheduling/annotation/AsyncExecutionTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,10 +16,12 @@
 
 package org.springframework.scheduling.annotation;
 
+import java.io.IOException;
 import java.io.Serializable;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.util.HashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
 import org.aopalliance.intercept.MethodInterceptor;
@@ -61,6 +63,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("autoProxyCreator", new RootBeanDefinition(DefaultAdvisorAutoProxyCreator.class));
 		context.registerBeanDefinition("asyncAdvisor", new RootBeanDefinition(AsyncAnnotationAdvisor.class));
 		context.refresh();
+
 		AsyncMethodBean asyncTest = context.getBean("asyncTest", AsyncMethodBean.class);
 		asyncTest.doNothing(5);
 		asyncTest.doSomething(10);
@@ -68,6 +71,24 @@ public class AsyncExecutionTests {
 		assertEquals("20", future.get());
 		ListenableFuture<String> listenableFuture = asyncTest.returnSomethingListenable(20);
 		assertEquals("20", listenableFuture.get());
+
+		future = asyncTest.returnSomething(0);
+		try {
+			future.get();
+			fail("Should have thrown ExecutionException");
+		}
+		catch (ExecutionException ex) {
+			assertTrue(ex.getCause() instanceof IllegalArgumentException);
+		}
+
+		future = asyncTest.returnSomething(-1);
+		try {
+			future.get();
+			fail("Should have thrown ExecutionException");
+		}
+		catch (ExecutionException ex) {
+			assertTrue(ex.getCause() instanceof IOException);
+		}
 	}
 
 	@Test
@@ -96,6 +117,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("e1", new RootBeanDefinition(ThreadPoolTaskExecutor.class));
 		context.registerBeanDefinition("e2", new RootBeanDefinition(ThreadPoolTaskExecutor.class));
 		context.refresh();
+
 		AsyncMethodWithQualifierBean asyncTest = context.getBean("asyncTest", AsyncMethodWithQualifierBean.class);
 		asyncTest.doNothing(5);
 		asyncTest.doSomething(10);
@@ -116,6 +138,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("e1", new RootBeanDefinition(ThreadPoolTaskExecutor.class));
 		context.registerBeanDefinition("e2", new RootBeanDefinition(ThreadPoolTaskExecutor.class));
 		context.refresh();
+
 		SimpleInterface asyncTest = context.getBean("asyncTest", SimpleInterface.class);
 		asyncTest.doNothing(5);
 		asyncTest.doSomething(10);
@@ -133,6 +156,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("autoProxyCreator", new RootBeanDefinition(DefaultAdvisorAutoProxyCreator.class));
 		context.registerBeanDefinition("asyncAdvisor", new RootBeanDefinition(AsyncAnnotationAdvisor.class));
 		context.refresh();
+
 		AsyncClassBean asyncTest = context.getBean("asyncTest", AsyncClassBean.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -148,6 +172,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("asyncTest", new RootBeanDefinition(AsyncClassBean.class));
 		context.registerBeanDefinition("asyncProcessor", new RootBeanDefinition(AsyncAnnotationBeanPostProcessor.class));
 		context.refresh();
+
 		AsyncClassBean asyncTest = context.getBean("asyncTest", AsyncClassBean.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -162,6 +187,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("autoProxyCreator", new RootBeanDefinition(DefaultAdvisorAutoProxyCreator.class));
 		context.registerBeanDefinition("asyncAdvisor", new RootBeanDefinition(AsyncAnnotationAdvisor.class));
 		context.refresh();
+
 		RegularInterface asyncTest = context.getBean("asyncTest", RegularInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -175,6 +201,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("asyncTest", new RootBeanDefinition(AsyncClassBeanWithInterface.class));
 		context.registerBeanDefinition("asyncProcessor", new RootBeanDefinition(AsyncAnnotationBeanPostProcessor.class));
 		context.refresh();
+
 		RegularInterface asyncTest = context.getBean("asyncTest", RegularInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -189,6 +216,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("autoProxyCreator", new RootBeanDefinition(DefaultAdvisorAutoProxyCreator.class));
 		context.registerBeanDefinition("asyncAdvisor", new RootBeanDefinition(AsyncAnnotationAdvisor.class));
 		context.refresh();
+
 		AsyncInterface asyncTest = context.getBean("asyncTest", AsyncInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -202,6 +230,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("asyncTest", new RootBeanDefinition(AsyncInterfaceBean.class));
 		context.registerBeanDefinition("asyncProcessor", new RootBeanDefinition(AsyncAnnotationBeanPostProcessor.class));
 		context.refresh();
+
 		AsyncInterface asyncTest = context.getBean("asyncTest", AsyncInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -216,6 +245,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("autoProxyCreator", new RootBeanDefinition(DefaultAdvisorAutoProxyCreator.class));
 		context.registerBeanDefinition("asyncAdvisor", new RootBeanDefinition(AsyncAnnotationAdvisor.class));
 		context.refresh();
+
 		AsyncInterface asyncTest = context.getBean("asyncTest", AsyncInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -229,6 +259,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("asyncTest", new RootBeanDefinition(DynamicAsyncInterfaceBean.class));
 		context.registerBeanDefinition("asyncProcessor", new RootBeanDefinition(AsyncAnnotationBeanPostProcessor.class));
 		context.refresh();
+
 		AsyncInterface asyncTest = context.getBean("asyncTest", AsyncInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -243,6 +274,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("autoProxyCreator", new RootBeanDefinition(DefaultAdvisorAutoProxyCreator.class));
 		context.registerBeanDefinition("asyncAdvisor", new RootBeanDefinition(AsyncAnnotationAdvisor.class));
 		context.refresh();
+
 		AsyncMethodsInterface asyncTest = context.getBean("asyncTest", AsyncMethodsInterface.class);
 		asyncTest.doNothing(5);
 		asyncTest.doSomething(10);
@@ -257,6 +289,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("asyncTest", new RootBeanDefinition(AsyncMethodsInterfaceBean.class));
 		context.registerBeanDefinition("asyncProcessor", new RootBeanDefinition(AsyncAnnotationBeanPostProcessor.class));
 		context.refresh();
+
 		AsyncMethodsInterface asyncTest = context.getBean("asyncTest", AsyncMethodsInterface.class);
 		asyncTest.doNothing(5);
 		asyncTest.doSomething(10);
@@ -271,6 +304,7 @@ public class AsyncExecutionTests {
 		context.registerBeanDefinition("asyncTest", new RootBeanDefinition(DynamicAsyncMethodsInterfaceBean.class));
 		context.registerBeanDefinition("asyncProcessor", new RootBeanDefinition(AsyncAnnotationBeanPostProcessor.class));
 		context.refresh();
+
 		AsyncMethodsInterface asyncTest = context.getBean("asyncTest", AsyncMethodsInterface.class);
 		asyncTest.doSomething(10);
 		Future<String> future = asyncTest.returnSomething(20);
@@ -351,7 +385,13 @@ public class AsyncExecutionTests {
 		@Async
 		public Future<String> returnSomething(int i) {
 			assertTrue(!Thread.currentThread().getName().equals(originalThreadName));
-			return new AsyncResult<String>(Integer.toString(i));
+			if (i == 0) {
+				throw new IllegalArgumentException();
+			}
+			else if (i < 0) {
+				return AsyncResult.forExecutionException(new IOException());
+			}
+			return AsyncResult.forValue(Integer.toString(i));
 		}
 
 		@Async