diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
index 70686f844b..7e4005797a 100644
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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. You may obtain a copy of the License at
@@ -31,6 +31,7 @@ import javax.sql.DataSource;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
@@ -75,6 +76,7 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @author Matt Stine
  * @author Gunnar Hillert
+ * @author Will Schipp
  *
  * @since 2.0
  */
@@ -419,10 +421,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 		List<Message<?>> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper);
 
-		if (messages.size() == 0){
-			return new SimpleMessageGroup(groupId);
-		}
-
 		jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key},
 				new RowCallbackHandler() {
 					public void processRow(ResultSet rs) throws SQLException {
diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
index 2fd4473c71..f6a7864663 100644
--- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
+++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -65,6 +65,7 @@ import org.springframework.transaction.annotation.Transactional;
  * @author Oleg Zhurakousky
  * @author Gunnar Hillert
  * @author Artem Bilan
+ * @author Will Schipp
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -424,9 +425,49 @@ public class JdbcMessageStoreTests {
 		LOG.info("messageFromGroup1: " + messageFromGroup1.getHeaders().getId() + "; Sequence #: " + messageFromGroup1.getHeaders().getSequenceNumber());
 		LOG.info("messageFromGroup2: " + messageFromGroup2.getHeaders().getId() + "; Sequence #: " + messageFromGroup2.getHeaders().getSequenceNumber());
 
-		assertEquals(Integer.valueOf(1), (Integer) messageFromGroup1.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
-		assertEquals(Integer.valueOf(2), (Integer) messageFromGroup2.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
+		assertEquals(Integer.valueOf(1), messageFromGroup1.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
+		assertEquals(Integer.valueOf(2), messageFromGroup2.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
 
 	}
 
+	@Test
+	@Transactional
+	public void testCompletedNotExpiredGroupINT3037() throws Exception {
+		/*
+		 * based on the aggregator scenario as follows;
+		 *
+		 * send three messages in
+		 * 1 of 2
+		 * 2 of 2
+		 * 2 of 2 (last again)
+		 *
+		 * expected behavior is that the LAST message (2 of 2 repeat) should be on the discard channel
+		 * (discard behavior performed by the AbstractCorrelatingMessageHandler.handleMessageInternal)
+		 */
+		final JdbcMessageStore messageStore = new JdbcMessageStore(dataSource);
+		//init
+		String groupId = "group";
+		//build the messages
+		Message<?> oneOfTwo = MessageBuilder.withPayload("hello").setSequenceNumber(1).setSequenceSize(2).setCorrelationId(groupId).build();
+		Message<?> twoOfTwo = MessageBuilder.withPayload("world").setSequenceNumber(2).setSequenceSize(2).setCorrelationId(groupId).build();
+		//add to the messageStore
+		messageStore.addMessageToGroup(groupId, oneOfTwo);
+		messageStore.addMessageToGroup(groupId, twoOfTwo);
+		//check that 2 messages are there
+		assertTrue(messageStore.getMessageGroupCount() == 1);
+		assertTrue(messageStore.getMessageCount() == 2);
+		//retrieve the group (like in the aggregator)
+		MessageGroup messageGroup = messageStore.getMessageGroup(groupId);
+		//'complete' the group
+		messageStore.completeGroup(messageGroup.getGroupId());
+		//now clear the messages
+		for (Message<?> message : messageGroup.getMessages()) {
+			messageStore.removeMessageFromGroup(groupId, message);
+		}//end for
+		//'add' the other message --> emulated by getting the messageGroup
+		messageGroup = messageStore.getMessageGroup(groupId);
+		//should be marked 'complete' --> old behavior it would not
+		assertTrue(messageGroup.isComplete());
+	}
+
 }