diff --git a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/core/JpaExecutor.java b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/core/JpaExecutor.java
index a04db2df47..0a3c9d5872 100644
--- a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/core/JpaExecutor.java
+++ b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/core/JpaExecutor.java
@@ -52,6 +52,7 @@ import org.springframework.util.Assert;
  * is "guessed" from the {@link Message} payload.
  *
  * @author Gunnar Hillert
+ * @author Amol Nayak
  * @since 2.2
  *
  */
@@ -143,7 +144,7 @@ public class JpaExecutor implements InitializingBean {
 	 */
 	public void afterPropertiesSet() {
 
-		if (this.jpaParameters != null ) {
+		if (this.jpaParameters != null) {
 
 			if (this.parameterSourceFactory == null) {
 				ExpressionEvaluatingParameterSourceFactory expressionSourceFactory =
@@ -151,7 +152,8 @@ public class JpaExecutor implements InitializingBean {
 				expressionSourceFactory.setParameters(jpaParameters);
 				this.parameterSourceFactory = expressionSourceFactory;
 
-			} else {
+			}
+			else {
 
 				if (!(this.parameterSourceFactory instanceof ExpressionEvaluatingParameterSourceFactory)) {
 					throw new IllegalStateException("You are providing 'JpaParameters'. "
@@ -166,7 +168,8 @@ public class JpaExecutor implements InitializingBean {
 				this.usePayloadAsParameterSource = false;
 			}
 
-		} else {
+		}
+		else {
 
 			if (this.parameterSourceFactory == null) {
 				this.parameterSourceFactory = new BeanPropertyParameterSourceFactory();
@@ -195,30 +198,40 @@ public class JpaExecutor implements InitializingBean {
 
 		final Object result;
 
+		ParameterSource parameterSource = null;
+		if (this.jpaQuery != null || this.nativeQuery != null || this.namedQuery != null) {
+			parameterSource = determineParameterSource(message);
+		}
 		if (this.jpaQuery != null) {
 
-			result = this.jpaOperations.executeUpdate(this.jpaQuery, parameterSourceFactory.createParameterSource(message));
+			result = this.jpaOperations.executeUpdate(this.jpaQuery, parameterSource);
 
-		} else if (this.nativeQuery != null) {
+		}
+		else if (this.nativeQuery != null) {
 
-			result = this.jpaOperations.executeUpdateWithNativeQuery(this.nativeQuery, parameterSourceFactory.createParameterSource(message));
+			result = this.jpaOperations.executeUpdateWithNativeQuery(this.nativeQuery, parameterSource);
 
-		} else if (this.namedQuery != null) {
+		}
+		else if (this.namedQuery != null) {
 
-			result = this.jpaOperations.executeUpdateWithNamedQuery(this.namedQuery, parameterSourceFactory.createParameterSource(message));
+			result = this.jpaOperations.executeUpdateWithNamedQuery(this.namedQuery, parameterSource);
 
-		} else {
+		}
+		else {
 
 			if (PersistMode.PERSIST.equals(this.persistMode)) {
 				this.jpaOperations.persist(message.getPayload());
 				result = message.getPayload();
-			} else if (PersistMode.MERGE.equals(this.persistMode)) {
+			}
+			else if (PersistMode.MERGE.equals(this.persistMode)) {
 				final Object mergedEntity = this.jpaOperations.merge(message.getPayload());
 				result = mergedEntity;
-			} else if (PersistMode.DELETE.equals(this.persistMode)) {
+			}
+			else if (PersistMode.DELETE.equals(this.persistMode)) {
 				this.jpaOperations.delete(message.getPayload());
 				result = message.getPayload();
-			} else {
+			}
+			else {
 				throw new IllegalStateException(String.format("Unsupported PersistMode: '%s'", this.persistMode.name()));
 			}
 
@@ -231,7 +244,7 @@ public class JpaExecutor implements InitializingBean {
 	/**
 	 * Execute a (typically retrieving) JPA operation. The <i>requestMessage</i>
 	 * can be used to provide additional query parameters using
-	 * {@link JpaExecutor#parameterSourceFactorymeterSourceFactory}. If the
+	 * {@link JpaExecutor#parameterSourceFactory}. If the
 	 * <i>requestMessage</i> parameter is null then
 	 * {@link JpaExecutor#parameterSource} is being used for providing query parameters.
 	 *
@@ -247,25 +260,31 @@ public class JpaExecutor implements InitializingBean {
 
 		if (requestMessage == null) {
 			result = doPoll(this.parameterSource);
-		} else {
-			result = doPoll(this.parameterSourceFactory.createParameterSource(requestMessage));
+		}
+		else {
+			ParameterSource parameterSource = determineParameterSource(requestMessage);
+
+			result = doPoll(parameterSource);
 		}
 
 		if (result.isEmpty()) {
 			payload = null;
-		} else {
+		}
+		else {
 
 			if (this.expectSingleResult) {
 				if (result.size() == 1) {
 					payload = result.iterator().next();
-				} else {
+				}
+				else {
 
 					throw new MessageHandlingException(requestMessage,
 						"The Jpa operation returned more than "
 					  + "1 result object but expectSingleResult was 'true'.");
 				}
 
-			} else {
+			}
+			else {
 				payload = result;
 			}
 
@@ -278,10 +297,12 @@ public class JpaExecutor implements InitializingBean {
 					for (Object entity : (Iterable<?>) payload) {
 						this.jpaOperations.delete(entity);
 					}
-				} else {
+				}
+				else {
 					this.jpaOperations.deleteInBatch((Iterable<Object>) payload);
 				}
-			} else {
+			}
+			else {
 				this.jpaOperations.delete(payload);
 			}
 
@@ -290,6 +311,17 @@ public class JpaExecutor implements InitializingBean {
 		return payload;
 	}
 
+	private ParameterSource determineParameterSource(final Message<?> requestMessage) {
+		ParameterSource parameterSource;
+		if (usePayloadAsParameterSource) {
+			parameterSource = this.parameterSourceFactory.createParameterSource(requestMessage.getPayload());
+		}
+		else {
+			parameterSource = this.parameterSourceFactory.createParameterSource(requestMessage);
+		}
+		return parameterSource;
+	}
+
 	/**
 	 * Execute the JPA operation. Delegates to {@link JpaExecutor#poll(Message)}.
 	 */
@@ -303,13 +335,17 @@ public class JpaExecutor implements InitializingBean {
 
 		if (this.jpaQuery != null) {
 			payload = jpaOperations.getResultListForQuery(this.jpaQuery, jpaQLParameterSource, maxNumberOfResults);
-		} else if (this.nativeQuery != null) {
+		}
+		else if (this.nativeQuery != null) {
 			payload = jpaOperations.getResultListForNativeQuery(this.nativeQuery, this.entityClass, jpaQLParameterSource, maxNumberOfResults);
-		} else if (this.namedQuery != null) {
+		}
+		else if (this.namedQuery != null) {
 			payload = jpaOperations.getResultListForNamedQuery(this.namedQuery, jpaQLParameterSource, maxNumberOfResults);
-		} else if (this.entityClass != null) {
+		}
+		else if (this.entityClass != null) {
 			payload = jpaOperations.getResultListForClass(this.entityClass, maxNumberOfResults);
-		} else {
+		}
+		else {
 			throw new IllegalStateException("For the polling operation, one of "
 								+ "the following properties must be specified: "
 								+ "query, namedQuery or entityClass.");
@@ -335,14 +371,9 @@ public class JpaExecutor implements InitializingBean {
 	 * @param jpaQuery The provided JPA query must neither be null nor empty.
 	 */
 	public void setJpaQuery(String jpaQuery) {
-
-		if (this.nativeQuery != null || this.namedQuery != null) {
-			throw new IllegalArgumentException("You can define only one of the " +
-							"properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
-		}
-
+		Assert.isTrue(this.nativeQuery == null && this.namedQuery == null, "You can define only one of the "
+							+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
 		Assert.hasText(jpaQuery, "jpaQuery must neither be null nor empty.");
-
 		this.jpaQuery = jpaQuery;
 	}
 
@@ -356,11 +387,8 @@ public class JpaExecutor implements InitializingBean {
 	 */
 	public void setNativeQuery(String nativeQuery) {
 
-		if (this.jpaQuery != null || this.namedQuery != null) {
-			throw new IllegalArgumentException("You can define only one of the " +
-							"properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
-		}
-
+		Assert.isTrue(this.namedQuery == null && this.jpaQuery == null, "You can define only one of the "
+				+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'");;
 		Assert.hasText(nativeQuery, "nativeQuery must neither be null nor empty.");
 
 		this.nativeQuery = nativeQuery;
@@ -374,10 +402,8 @@ public class JpaExecutor implements InitializingBean {
 	 */
 	public void setNamedQuery(String namedQuery) {
 
-		if (this.jpaQuery != null || this.nativeQuery != null) {
-			throw new IllegalArgumentException("You can define only one of the " +
-							"properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
-		}
+		Assert.isTrue(this.jpaQuery == null && this.nativeQuery == null, "You can define only one of the "
+				+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
 
 		Assert.hasText(namedQuery, "namedQuery must neither be null nor empty.");
 		this.namedQuery = namedQuery;
@@ -450,4 +476,29 @@ public class JpaExecutor implements InitializingBean {
 		this.expectSingleResult = expectSingleResult;
 	}
 
+	//Exposing getters for the unit test cases
+
+	/**
+	 * Returns the JPA Query that would be executed using the executor
+	 */
+	public String getJpaQuery() {
+		return jpaQuery;
+	}
+
+	/**
+	 * Returns the Native Query that would be executed using the executor
+	 */
+	public String getNativeQuery() {
+		return nativeQuery;
+	}
+
+	/**
+	 * Returns the Named Query that would be executed using the executor
+	 */
+	public String getNamedQuery() {
+		return namedQuery;
+	}
+
+
+
 }
diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests-context.xml b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests-context.xml
new file mode 100644
index 0000000000..12c9a811ba
--- /dev/null
+++ b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests-context.xml
@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:util="http://www.springframework.org/schema/util"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
+	
+	<import resource="classpath:/hibernateJpa-context.xml" />
+
+</beans>
\ No newline at end of file
diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests.java b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests.java
index 82ff68f051..54c6578b19 100644
--- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests.java
+++ b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/JpaExecutorTests.java
@@ -14,20 +14,39 @@ package org.springframework.integration.jpa.core;
 
 import static org.mockito.Mockito.mock;
 
+import java.util.Collections;
+import java.util.Map;
+
 import javax.persistence.EntityManager;
 
 import junit.framework.Assert;
 
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.Message;
+import org.springframework.integration.jpa.support.JpaParameter;
+import org.springframework.integration.jpa.support.parametersource.ExpressionEvaluatingParameterSourceFactory;
+import org.springframework.integration.jpa.test.entity.StudentDomain;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.annotation.Transactional;
 
 /**
  *
  * @author Gunnar Hillert
+ * @author Amol Nayak
  * @since 2.2
  *
  */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
 public class JpaExecutorTests {
 
+	@Autowired
+	protected EntityManager entityManager;
+
 	/**
 	 * In this test, the {@link JpaExecutor}'s poll method will be called without
 	 * specifying a 'query', 'namedQuery' or 'entityClass' property. This should
@@ -69,4 +88,122 @@ public class JpaExecutorTests {
 
 	}
 
-}
+	@Test
+	public void testSetMultipleQueryTypes() {
+		JpaExecutor executor = new JpaExecutor(mock(EntityManager.class));
+		executor.setJpaQuery("select s from Student s");
+		Assert.assertNotNull(executor.getJpaQuery());
+
+		try {
+			executor.setNamedQuery("NamedQuery");
+		} catch (IllegalArgumentException e) {
+			Assert.assertEquals("You can define only one of the "
+				+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'", e.getMessage());
+		}
+		Assert.assertNull(executor.getNamedQuery());
+
+		try {
+			executor.setNativeQuery("select * from Student");
+		} catch (IllegalArgumentException e) {
+			Assert.assertEquals("You can define only one of the "
+					+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'", e.getMessage());
+		}
+		Assert.assertNull(executor.getNativeQuery());
+
+		executor = new JpaExecutor(mock(EntityManager.class));
+		executor.setNamedQuery("NamedQuery");
+		Assert.assertNotNull(executor.getNamedQuery());
+
+		try {
+			executor.setJpaQuery("select s from Student s");
+		} catch (IllegalArgumentException e) {
+			Assert.assertEquals("You can define only one of the "
+					+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'", e.getMessage());
+		}
+		Assert.assertNull(executor.getJpaQuery());
+
+	}
+
+	@Test
+	@Transactional
+	public void selectWithMessageAsParameterSource() {
+		String query = "select s from Student s where s.firstName = :firstName";
+		Message<Map<String, String>> message =
+			MessageBuilder.withPayload(Collections.singletonMap("firstName", "First One")).build();
+		JpaExecutor executor = getJpaExecutorForMessageAsParamSource(query);
+		StudentDomain student = (StudentDomain) executor.poll(message);
+		Assert.assertNotNull(student);
+	}
+
+	@Test
+	@Transactional
+	public void selectWithPayloadAsParameterSource() {
+		String query = "select s from Student s where s.firstName = :firstName";
+		Message<String> message =
+			MessageBuilder.withPayload("First One").build();
+		JpaExecutor executor = getJpaExecutorForPayloadAsParamSource(query);
+		StudentDomain student = (StudentDomain) executor.poll(message);
+		Assert.assertNotNull(student);
+	}
+
+	@Test
+	@Transactional
+	public void updateWithMessageAsParameterSource() {
+		String query = "update Student s set s.firstName = :firstName where s.lastName = 'Last One'";
+		Message<Map<String, String>> message =
+			MessageBuilder.withPayload(Collections.singletonMap("firstName", "First One")).build();
+		JpaExecutor executor = getJpaExecutorForMessageAsParamSource(query);
+		Integer rowsAffected = (Integer) executor.executeOutboundJpaOperation(message);
+		Assert.assertTrue(1 == rowsAffected);
+	}
+
+	@Test
+	@Transactional
+	public void updateWithPayloadAsParameterSource() {
+		String query = "update Student s set s.firstName = :firstName where s.lastName = 'Last One'";
+		Message<String> message =
+			MessageBuilder.withPayload("First One").build();
+		JpaExecutor executor = getJpaExecutorForPayloadAsParamSource(query);
+		Integer rowsAffected = (Integer) executor.executeOutboundJpaOperation(message);
+		Assert.assertTrue(1 == rowsAffected);
+	}
+
+	/**
+	 * @param query
+	 * @return
+	 */
+	private JpaExecutor getJpaExecutorForMessageAsParamSource(String query) {
+		JpaExecutor executor = new JpaExecutor(entityManager);
+		ExpressionEvaluatingParameterSourceFactory factory =
+				new ExpressionEvaluatingParameterSourceFactory();
+		factory.setParameters(
+				Collections.singletonList(new JpaParameter("firstName", null, "payload['firstName']")));
+		executor.setParameterSourceFactory(factory);
+		executor.setJpaQuery(query);
+		executor.setExpectSingleResult(true);
+		executor.setUsePayloadAsParameterSource(false);
+		executor.afterPropertiesSet();
+		return executor;
+	}
+
+	/**
+	 * @param query
+	 * @return
+	 */
+	private JpaExecutor getJpaExecutorForPayloadAsParamSource(String query) {
+		JpaExecutor executor = new JpaExecutor(entityManager);
+		ExpressionEvaluatingParameterSourceFactory factory =
+				new ExpressionEvaluatingParameterSourceFactory();
+		factory.setParameters(
+				Collections.singletonList(new JpaParameter("firstName", null, "#this")));
+		executor.setParameterSourceFactory(factory);
+		executor.setJpaQuery(query);
+		executor.setExpectSingleResult(true);
+		executor.setUsePayloadAsParameterSource(true);
+		executor.afterPropertiesSet();
+		return executor;
+	}
+
+
+
+}
\ No newline at end of file