diff --git a/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParser.java b/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParser.java
index d2f1a9a0..55918341 100644
--- a/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParser.java
+++ b/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParser.java
@@ -62,6 +62,7 @@ class FlowExecutorBeanDefinitionParser extends AbstractSingleBeanDefinitionParse
 		Element repositoryElement = DomUtils.getChildElementByTagName(element, "flow-execution-repository");
 		if (repositoryElement != null) {
 			addMaxExecutions(repositoryElement, definitionBuilder, parserContext);
+			addConversationManager(repositoryElement, definitionBuilder, parserContext);
 			addMaxSnapshots(repositoryElement, definitionBuilder, parserContext);
 		}
 	}
@@ -73,6 +74,14 @@ class FlowExecutorBeanDefinitionParser extends AbstractSingleBeanDefinitionParse
 		}
 	}
 
+	private void addConversationManager(Element element, BeanDefinitionBuilder definitionBuilder,
+			ParserContext parserContext) {
+		String conversationManager = element.getAttribute("conversation-manager");
+		if (StringUtils.hasText(conversationManager)) {
+			definitionBuilder.addPropertyReference("conversationManager", conversationManager);
+		}
+	}
+
 	private void addMaxSnapshots(Element element, BeanDefinitionBuilder definitionBuilder, ParserContext parserContext) {
 		String maxSnapshots = element.getAttribute("max-execution-snapshots");
 		if (StringUtils.hasText(maxSnapshots)) {
diff --git a/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorFactoryBean.java b/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorFactoryBean.java
index 177a6991..a4f7a7b5 100644
--- a/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorFactoryBean.java
+++ b/spring-webflow/src/main/java/org/springframework/webflow/config/FlowExecutorFactoryBean.java
@@ -72,6 +72,8 @@ class FlowExecutorFactoryBean implements FactoryBean, ApplicationContextAware, B
 
 	private FlowExecutionListenerLoader flowExecutionListenerLoader;
 
+	private ConversationManager conversationManager;
+
 	private ConversionService conversionService;
 
 	private FlowExecutor flowExecutor;
@@ -121,6 +123,14 @@ class FlowExecutorFactoryBean implements FactoryBean, ApplicationContextAware, B
 		this.flowExecutionListenerLoader = flowExecutionListenerLoader;
 	}
 
+	/**
+	 * Sets the service type that manages conversations and effectively controls how state is stored physically when a
+	 * flow execution is paused.
+	 */
+	public void setConversationManager(ConversationManager conversationManager) {
+		this.conversationManager = conversationManager;
+	}
+
 	// implementing ApplicationContextAware
 
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -194,11 +204,14 @@ class FlowExecutorFactoryBean implements FactoryBean, ApplicationContextAware, B
 	}
 
 	private ConversationManager createConversationManager() {
-		SessionBindingConversationManager conversationManager = new SessionBindingConversationManager();
-		if (maxFlowExecutions != null) {
-			conversationManager.setMaxConversations(maxFlowExecutions.intValue());
+		if (conversationManager == null) {
+			conversationManager = new SessionBindingConversationManager();
+			if (maxFlowExecutions != null) {
+				((SessionBindingConversationManager) conversationManager).setMaxConversations(maxFlowExecutions
+						.intValue());
+			}
 		}
-		return conversationManager;
+		return this.conversationManager;
 	}
 
 	private FlowExecutionSnapshotFactory createFlowExecutionSnapshotFactory(FlowExecutionFactory executionFactory) {
diff --git a/spring-webflow/src/main/java/org/springframework/webflow/config/spring-webflow-config-2.3.xsd b/spring-webflow/src/main/java/org/springframework/webflow/config/spring-webflow-config-2.3.xsd
index 8011d1c5..f187db5c 100644
--- a/spring-webflow/src/main/java/org/springframework/webflow/config/spring-webflow-config-2.3.xsd
+++ b/spring-webflow/src/main/java/org/springframework/webflow/config/spring-webflow-config-2.3.xsd
@@ -3,6 +3,7 @@
 	xmlns="http://www.springframework.org/schema/webflow-config"
 	xmlns:xsd="http://www.w3.org/2001/XMLSchema"
 	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:tool="http://www.springframework.org/schema/tool"
 	targetNamespace="http://www.springframework.org/schema/webflow-config"
 	elementFormDefault="qualified" attributeFormDefault="unqualified"
 	version="2.3">
@@ -19,6 +20,7 @@ A XML-based DSL for configuring the Spring Web Flow 2.0 system.
 	</xsd:annotation>
 
 	<xsd:import namespace="http://www.springframework.org/schema/beans" schemaLocation="http://www.springframework.org/schema/beans/spring-beans-3.0.xsd" />
+	<xsd:import namespace="http://www.springframework.org/schema/tool" schemaLocation="http://www.springframework.org/schema/tool/spring-tool-3.0.xsd" />
 
 	<xsd:element name="flow-registry">
 		<xsd:annotation>
@@ -323,11 +325,27 @@ The maximum number of persistent flow executions allowed per user session.  The
 				<xsd:documentation>
 				<![CDATA[
 The maximum number of history snapshots allowed per flow execution.  The default is 30. 
-Setting this value to 0 disables the creation of snapshot copies altogether.  
+Setting the value to 0 disables the creation of snapshot copies altogether.
+Note that this attribute is meaningless when a custom ConversationManager is provided.
 ]]>
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="conversation-manager" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation source="java:org.springframework.webflow.conversation.ConversationManager"><![CDATA[
+The bean name of the ConversationManager to use to manage conversations.
+The default implementation stores conversations in the session.
+Note that when this attribute is provided, the "max-execution-snapshots" attribute is meaningless. 
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="java:org.springframework.webflow.conversation.ConversationManager" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		
 	</xsd:complexType>
 
 	<xsd:complexType name="flowExecutionListenersType">
diff --git a/spring-webflow/src/test/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParserTests.java b/spring-webflow/src/test/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParserTests.java
index 972a0f9d..7942af48 100644
--- a/spring-webflow/src/test/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParserTests.java
+++ b/spring-webflow/src/test/java/org/springframework/webflow/config/FlowExecutorBeanDefinitionParserTests.java
@@ -3,10 +3,16 @@ package org.springframework.webflow.config;
 import junit.framework.TestCase;
 
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.webflow.conversation.Conversation;
+import org.springframework.webflow.conversation.ConversationException;
+import org.springframework.webflow.conversation.ConversationId;
+import org.springframework.webflow.conversation.ConversationManager;
+import org.springframework.webflow.conversation.ConversationParameters;
 import org.springframework.webflow.definition.FlowDefinition;
 import org.springframework.webflow.execution.FlowExecutionListenerAdapter;
 import org.springframework.webflow.execution.RequestContext;
 import org.springframework.webflow.executor.FlowExecutor;
+import org.springframework.webflow.executor.FlowExecutorImpl;
 import org.springframework.webflow.test.MockExternalContext;
 
 public class FlowExecutorBeanDefinitionParserTests extends TestCase {
@@ -17,22 +23,46 @@ public class FlowExecutorBeanDefinitionParserTests extends TestCase {
 	}
 
 	public void testConfigOk() {
-		FlowExecutor executor = (FlowExecutor) context.getBean("flowExecutor", FlowExecutor.class);
+		FlowExecutor executor = context.getBean("flowExecutor", FlowExecutor.class);
 		executor.launchExecution("flow", null, new MockExternalContext());
-		FlowExecutor executor2 = (FlowExecutor) context.getBean("flowExecutorSimpleRepo", FlowExecutor.class);
+		FlowExecutor executor2 = context.getBean("flowExecutorSimpleRepo", FlowExecutor.class);
 		executor2.launchExecution("flow", null, new MockExternalContext());
 	}
 
+	public void testCustomConversationManager() {
+		FlowExecutorImpl executor = context.getBean("flowExecutor", FlowExecutorImpl.class);
+		try {
+			executor.getExecutionRepository().parseFlowExecutionKey("e1s1");
+			fail("ExceptionThrowingConversationManager would have raised an exception");
+		} catch (UnsupportedOperationException e) {
+		}
+	}
+
 	public static class ConfigurationListener extends FlowExecutionListenerAdapter {
 		public void sessionCreating(RequestContext context, FlowDefinition definition) {
-			if (!context.getFlowExecutionContext().isActive()) {
-				assertEquals(3, context.getFlowExecutionContext().getAttributes().size());
-				assertEquals(Boolean.FALSE, context.getFlowExecutionContext().getAttributes().getBoolean(
-						"alwaysRedirectOnPause"));
-				assertEquals("bar", context.getFlowExecutionContext().getAttributes().get("foo"));
-				assertEquals(new Integer(2), context.getFlowExecutionContext().getAttributes().get("bar"));
-			}
+			assertEquals(3, context.getFlowExecutionContext().getAttributes().size());
+			assertEquals(Boolean.FALSE,
+					context.getFlowExecutionContext().getAttributes().getBoolean("alwaysRedirectOnPause"));
+			assertEquals("bar", context.getFlowExecutionContext().getAttributes().get("foo"));
+			assertEquals(new Integer(2), context.getFlowExecutionContext().getAttributes().get("bar"));
 		}
 	}
 
+	public static class ExceptionThrowingConversationManager implements ConversationManager {
+
+		public Conversation beginConversation(ConversationParameters conversationParameters)
+				throws ConversationException {
+			throw new UnsupportedOperationException();
+		}
+
+		public Conversation getConversation(ConversationId id) throws ConversationException {
+			throw new UnsupportedOperationException();
+		}
+
+		public ConversationId parseConversationId(String encodedId) throws ConversationException {
+			throw new UnsupportedOperationException();
+		}
+
+	}
+
 }
diff --git a/spring-webflow/src/test/java/org/springframework/webflow/config/flow-executor.xml b/spring-webflow/src/test/java/org/springframework/webflow/config/flow-executor.xml
index 76015639..e66f344e 100644
--- a/spring-webflow/src/test/java/org/springframework/webflow/config/flow-executor.xml
+++ b/spring-webflow/src/test/java/org/springframework/webflow/config/flow-executor.xml
@@ -9,7 +9,7 @@
            http://www.springframework.org/schema/webflow-config/spring-webflow-config.xsd">
 
 	<webflow:flow-executor id="flowExecutor" flow-registry="flowRegistry">
-		<webflow:flow-execution-repository max-executions="1" max-execution-snapshots="2"/>
+		<webflow:flow-execution-repository max-executions="1" max-execution-snapshots="2" conversation-manager="conversationManager" />
 		<webflow:flow-execution-attributes>
 			<webflow:always-redirect-on-pause value="false"/>
 			<webflow:attribute name="foo" value="bar"/>		
@@ -21,6 +21,8 @@
 	</webflow:flow-executor>
 	
 	<bean id="listener" class="org.springframework.webflow.config.FlowExecutorBeanDefinitionParserTests$ConfigurationListener" />
+
+	<bean id="conversationManager" class="org.springframework.webflow.config.FlowExecutorBeanDefinitionParserTests.ExceptionThrowingConversationManager" />
 	
 	<webflow:flow-registry id="flowRegistry">
 		<webflow:flow-location path="org/springframework/webflow/config/flow.xml" />