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 020ebe4d98..da7b60b001 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
@@ -89,6 +89,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 		REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"),
 
+		REMOVE_GROUP_TO_MESSAGE_JOIN("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"),
+
 		COUNT_ALL_MESSAGES_IN_GROUPS("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"),
 
 		COUNT_ALL_MESSAGES_IN_GROUP("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"),
@@ -403,7 +405,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 		List<Message<?>> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper);
 
-        jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key},
+		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 {
 						updateDate.set(rs.getTimestamp("UPDATED_DATE"));
@@ -416,11 +422,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 					}
 				});
 
-		if (messages.size() == 0){
+		if (createDate.get() == null && updateDate.get() == null) {
+			if (logger.isWarnEnabled()) {
+				for (Message<?> message : messages) {
+					logger.warn("Missing group row for message id: " + message.getHeaders().getId());
+				}
+			}
 			return new SimpleMessageGroup(groupId);
 		}
-		Assert.state(createDate.get() != null, "Could not locate created date for groupId=" + groupId);
-		Assert.state(updateDate.get() != null, "Could not locate updated date for groupId=" + groupId);
 
 		long timestamp = createDate.get().getTime();
 		boolean complete = completeFlag.get().booleanValue();
@@ -460,6 +469,15 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 			this.removeMessage(messageIds);
 		}
 
+		jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), new PreparedStatementSetter() {
+			public void setValues(PreparedStatement ps) throws SQLException {
+				if (logger.isDebugEnabled()){
+					logger.debug("Removing relationships for the group with group key=" + groupKey);
+				}
+				ps.setString(1, groupKey);
+			}
+		});
+
 		jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), new PreparedStatementSetter() {
 			public void setValues(PreparedStatement ps) throws SQLException {
 				if (logger.isDebugEnabled()){
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 eb13322028..cfc9b9143f 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
@@ -16,6 +16,15 @@
 
 package org.springframework.integration.jdbc;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
+
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
@@ -41,20 +50,12 @@ import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupCallback;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.util.UUIDConverter;
+import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.transaction.annotation.Transactional;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNotSame;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-
-import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
-
 /**
  * @author Dave Syer
  * @author Mark Fisher
@@ -234,6 +235,24 @@ public class JdbcMessageStoreTests {
 		assertEquals(0, group.size());
 	}
 
+	@Test
+	@Transactional
+	public void testRemoveMessageGroup() throws Exception {
+		JdbcTemplate template = new JdbcTemplate(dataSource);
+		template.afterPropertiesSet();
+		String groupId = "X";
+
+		Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
+		messageStore.addMessageToGroup(groupId, message);
+		messageStore.removeMessageGroup(groupId);
+		MessageGroup group = messageStore.getMessageGroup(groupId);
+		assertEquals(0, group.size());
+
+		String uuidGroupId = UUIDConverter.getUUID(groupId).toString();
+		assertTrue(template.queryForList(
+				"SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '"  + uuidGroupId + "'").size() == 0);
+	}
+
 	@Test
 	@Transactional
 	public void testCompleteMessageGroup() throws Exception {
@@ -279,7 +298,7 @@ public class JdbcMessageStoreTests {
 	@Transactional
 	public void testOrderInMessageGroup() throws Exception {
 		String groupId = "X";
-		
+
 		messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("foo").setCorrelationId(groupId).build());
 		Thread.sleep(1);
 		messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build());