diff --git a/org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java b/org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java
index 6abdb07aa9f..00bde493bb6 100644
--- a/org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java
+++ b/org.springframework.jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-2009 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.
@@ -28,6 +28,7 @@ import javax.jms.Topic;
 
 import org.springframework.jms.support.JmsUtils;
 import org.springframework.util.Assert;
+import org.springframework.util.ErrorHandler;
 
 /**
  * Abstract base class for message listener containers. Can either host
@@ -44,7 +45,11 @@ import org.springframework.util.Assert;
  * is to <b>never</b> propagate an exception thrown by a message listener up to
  * the JMS provider. Instead, it will log any such exception at the error level.
  * This means that from the perspective of the attendant JMS provider no such
- * listener will ever fail.
+ * listener will ever fail. However, if error handling is necessary, then
+ * any implementation of the {@link ErrorHandler} strategy may be provided to
+ * the {@link #setErrorHandler(ErrorHandler)} method. Note that JMSExceptions
+ * <b>will</b> be passed to the ErrorHandler in addition to (but after) being
+ * passed to an {@link ExceptionListener}, if one has been provided.
  *
  * <p>The listener container offers the following message acknowledgment options:
  * <ul>
@@ -132,6 +137,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 
 	private ExceptionListener exceptionListener;
 
+	private ErrorHandler errorHandler;
+
 	private boolean exposeListenerSession = true;
 
 	private boolean acceptMessagesWhileStopping = false;
@@ -342,6 +349,15 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		return this.exceptionListener;
 	}
 
+	/**
+	 * Set an ErrorHandler to be invoked in case of any uncaught exceptions thrown
+	 * while processing a Message. By default there will be <b>no</b> ErrorHandler
+	 * so that error-level logging is the only result.
+	 */
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
 	/**
 	 * Set whether to expose the listener JMS Session to a registered
 	 * {@link SessionAwareMessageListener} as well as to
@@ -642,8 +658,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		}
 		if (isActive()) {
 			// Regular case: failed while active.
-			// Log at error level.
-			logger.warn("Execution of JMS message listener failed", ex);
+			// Invoke ErrorHandler if available.
+			invokeErrorHandler(ex);
 		}
 		else {
 			// Rare case: listener thread failed after container shutdown.
@@ -664,6 +680,20 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		}
 	}
 
+	/**
+	 * Invoke the registered ErrorHandler, if any. Log at error level otherwise.
+	 * @param ex the uncaught error that arose during JMS processing.
+	 * @see #setErrorHandler
+	 */
+	protected void invokeErrorHandler(Throwable ex) {
+		if (this.errorHandler != null) {
+			this.errorHandler.handleError(ex);
+		}
+		else if (logger.isWarnEnabled()) {
+			logger.warn("Execution of JMS message listener failed, and no ErrorHandler has been set.", ex);
+		}
+	}
+
 
 	/**
 	 * Internal exception class that indicates a rejected message on shutdown.
diff --git a/org.springframework.jms/src/test/java/org/springframework/jms/listener/SimpleMessageListenerContainerTests.java b/org.springframework.jms/src/test/java/org/springframework/jms/listener/SimpleMessageListenerContainerTests.java
index a9781add764..453c461fe6d 100644
--- a/org.springframework.jms/src/test/java/org/springframework/jms/listener/SimpleMessageListenerContainerTests.java
+++ b/org.springframework.jms/src/test/java/org/springframework/jms/listener/SimpleMessageListenerContainerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-2009 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.
@@ -29,17 +29,20 @@ import javax.jms.MessageConsumer;
 import javax.jms.MessageListener;
 import javax.jms.Session;
 
+import org.easymock.EasyMock;
 import org.easymock.MockControl;
 import org.easymock.internal.AlwaysMatcher;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.jms.StubQueue;
+import org.springframework.util.ErrorHandler;
 
 /**
  * @author Rick Evans
  * @author Juergen Hoeller
  * @author Chris Beams
+ * @author Mark Fisher
  */
 public class SimpleMessageListenerContainerTests extends AbstractMessageListenerContainerTests {
 
@@ -380,6 +383,64 @@ public class SimpleMessageListenerContainerTests extends AbstractMessageListener
 		mockConnectionFactory.verify();
 	}
 
+	@Test
+	public void testRegisteredErrorHandlerIsInvokedOnException() throws Exception {
+		final SimpleMessageConsumer messageConsumer = new SimpleMessageConsumer();
+
+		Session session = EasyMock.createMock(Session.class);
+
+		// Queue gets created in order to create MessageConsumer for that Destination...
+		session.createQueue(DESTINATION_NAME);
+		EasyMock.expectLastCall().andReturn(QUEUE_DESTINATION);
+		// and then the MessageConsumer gets created...
+		session.createConsumer(QUEUE_DESTINATION, null); // no MessageSelector...
+		EasyMock.expectLastCall().andReturn(messageConsumer);
+		// an exception is thrown, so the rollback logic is being applied here...
+		session.getTransacted();
+		EasyMock.expectLastCall().andReturn(false);
+		EasyMock.replay(session);
+
+		Connection connection = EasyMock.createMock(Connection.class);
+		connection.setExceptionListener(this.container);
+		// session gets created in order to register MessageListener...
+		connection.createSession(this.container.isSessionTransacted(), this.container.getSessionAcknowledgeMode());
+		EasyMock.expectLastCall().andReturn(session);
+		// and the connection is start()ed after the listener is registered...
+		connection.start();
+		EasyMock.replay(connection);
+
+		ConnectionFactory connectionFactory = EasyMock.createMock(ConnectionFactory.class);
+		connectionFactory.createConnection();
+		EasyMock.expectLastCall().andReturn(connection);
+		EasyMock.replay(connectionFactory);
+
+		final IllegalStateException theException = new IllegalStateException("intentional test failure");
+
+		this.container.setConnectionFactory(connectionFactory);
+		this.container.setDestinationName(DESTINATION_NAME);
+		this.container.setMessageListener(new SessionAwareMessageListener() {
+			public void onMessage(Message message, Session session) throws JMSException {
+				throw theException;
+			}
+		});
+
+		ErrorHandler errorHandler = EasyMock.createMock(ErrorHandler.class);
+		errorHandler.handleError(theException);
+		EasyMock.expectLastCall();
+		EasyMock.replay(errorHandler);
+		this.container.setErrorHandler(errorHandler);
+		this.container.afterPropertiesSet();
+
+		// manually trigger an Exception with the above bad MessageListener...
+		Message message = EasyMock.createMock(Message.class);
+		EasyMock.replay(message);
+
+		// a Throwable from a MessageListener MUST simply be swallowed...
+		messageConsumer.sendMessage(message);
+
+		EasyMock.verify(errorHandler, message, session, connection, connectionFactory);
+	}
+
 	@Test
 	public void testNoRollbackOccursIfSessionIsNotTransactedAndThatExceptionsDo_NOT_Propagate() throws Exception {
 		final SimpleMessageConsumer messageConsumer = new SimpleMessageConsumer();