diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
index 0bd0dce470..9a4603f7e1 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
@@ -60,6 +60,22 @@
 			<xsd:attribute name="channel" type="xsd:string" use="required"/>
 			<xsd:attribute name="poll-period" type="xsd:int"/>
 			<xsd:attribute name="message-converter" type="xsd:string"/>
+			<xsd:attribute name="acknowledge" default="auto">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+	The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
+	The latter effectively activates a locally transacted Session.
+					]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:NMTOKEN">
+						<xsd:enumeration value="auto"/>
+						<xsd:enumeration value="client"/>
+						<xsd:enumeration value="dups-ok"/>
+						<xsd:enumeration value="transacted"/>
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
index 2eeeaf1337..2030bb1dd3 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
@@ -18,6 +18,7 @@ package org.springframework.integration.adapter.jms;
 
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
+import javax.jms.Session;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
@@ -37,29 +38,33 @@ import org.springframework.util.Assert;
  */
 public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object> implements Lifecycle, DisposableBean {
 
-	private AbstractJmsListeningContainer container;
+	private volatile AbstractJmsListeningContainer container;
 
-	private ConnectionFactory connectionFactory;
+	private volatile ConnectionFactory connectionFactory;
 
-	private Destination destination;
+	private volatile Destination destination;
 
-	private String destinationName;
+	private volatile String destinationName;
 
-	private MessageConverter messageConverter = new SimpleMessageConverter();
+	private volatile MessageConverter messageConverter = new SimpleMessageConverter();
 
-	private TaskExecutor taskExecutor;
+	private volatile TaskExecutor taskExecutor;
 
-	private long receiveTimeout = 1000;
+	private volatile boolean sessionTransacted;
 
-	private int concurrentConsumers = 1;
+	private volatile int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
 
-	private int maxConcurrentConsumers = 1;
+	private volatile long receiveTimeout = 1000;
 
-	private int maxMessagesPerTask = Integer.MIN_VALUE;
+	private volatile int concurrentConsumers = 1;
 
-	private int idleTaskExecutionLimit = 1;
+	private volatile int maxConcurrentConsumers = 1;
 
-	private long sendTimeout = -1;
+	private volatile int maxMessagesPerTask = Integer.MIN_VALUE;
+
+	private volatile int idleTaskExecutionLimit = 1;
+
+	private volatile long sendTimeout = -1;
 
 
 	public void setContainer(AbstractJmsListeningContainer container) {
@@ -91,6 +96,14 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
 		this.taskExecutor = taskExecutor;
 	}
 
+	public void setSessionTransacted(boolean sessionTransacted) {
+		this.sessionTransacted = sessionTransacted;
+	}
+
+	public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) {
+		this.sessionAcknowledgeMode = sessionAcknowledgeMode;
+	}
+
 	@Override
 	public void initialize() {
 		if (this.container == null) {
@@ -116,7 +129,8 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
 		dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
 		dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask);
 		dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit);
-		dmlc.setSessionTransacted(true);
+		dmlc.setSessionTransacted(this.sessionTransacted);
+		dmlc.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
 		dmlc.setAutoStartup(false);
 		ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel());
 		listener.setMessageConverter(this.messageConverter);
diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java
index 0510175c15..4741abf880 100644
--- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java
+++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.adapter.jms.config;
 
+import javax.jms.Session;
+
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.BeanCreationException;
@@ -58,6 +60,16 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
 
 	private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter";
 
+	private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge";
+
+	private static final String ACKNOWLEDGE_AUTO = "auto";
+
+	private static final String ACKNOWLEDGE_CLIENT = "client";
+
+	private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok";
+
+	private static final String ACKNOWLEDGE_TRANSACTED = "transacted";
+
 
 	protected Class<?> getBeanClass(Element element) {
 		if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) {
@@ -153,6 +165,39 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
 		if (StringUtils.hasText(messageConverter)) {
 			builder.addPropertyReference(MESSAGE_CONVERTER_PROPERTY, messageConverter);
 		}
+		Integer acknowledgeMode = parseAcknowledgeMode(element);
+		if (acknowledgeMode != null) {
+			if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) {
+				builder.addPropertyValue("sessionTransacted", Boolean.TRUE);
+			}
+			else {
+				builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode);
+			}
+		}
+	}
+
+	private Integer parseAcknowledgeMode(Element element) {
+		String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE);
+		if (StringUtils.hasText(acknowledge)) {
+			int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
+			if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) {
+				acknowledgeMode = Session.SESSION_TRANSACTED;
+			}
+			else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) {
+				acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
+			}
+			else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) {
+				acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
+			}
+			else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) {
+				throw new BeanCreationException("Invalid jms-source 'acknowledge' setting: " +
+						"only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.");
+			}
+			return acknowledgeMode;
+		}
+		else {
+			return null;
+		}
 	}
 
 }