INT-1339: Added support for partial completion for disordered sequences.

- Added markMessageFromGroup to MessageGroup interface
- made ResequencingMessageGroupProcessor return the partial sequence that was processed
- removed obsolete test
- refactored CorrelatingMessageHandler to deal with the partially processed sequence properly

Still it is needed to remove the release-strategy attribute from the namespace to avoid confusion.

spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java

@@ -27,6 +27,7 @@ import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.store.*;
 import org.springframework.util.Assert;
 
+import java.util.Collection;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -48,16 +49,13 @@ import java.util.concurrent.ConcurrentMap;
  * @author Dave Syer
  * @since 2.0
  */
+@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter"})
 public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer {
 
 	private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class);
 
 	public static final long DEFAULT_SEND_TIMEOUT = 1000L;
 
-	public static final long DEFAULT_REAPER_INTERVAL = 1000L;
-
-	public static final long DEFAULT_TIMEOUT = 60000L;
-
 
 	private MessageGroupStore messageStore;
 
@@ -166,21 +164,14 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 			if (group.canAdd(message)) {
 				group = store(correlationKey, message);
 				if (releaseStrategy.canRelease(group)) {
+					Collection<Message> completedMessages = null;
 					try {
-						completeGroup(message, correlationKey, group);
+						completedMessages = completeGroup(message, correlationKey, group);
 					}
 					finally {
 						// Always clean up even if there was an exception
 						// processing messages
-						if (group.isComplete() || group.getSequenceSize() == 0) {
-							// The group is complete or else there is no
-							// sequence so there is no more state to track
-							remove(group);
-						} else {
-							// Mark these messages as processed, but do not
-							// remove the group from store
-							mark(group);
-						}
+						cleanUpForReleasedGroup(group, completedMessages);
 					}
 				} else if (group.isComplete()) {
 					try {
@@ -200,6 +191,22 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	private void cleanUpForReleasedGroup(MessageGroup group, Collection<Message> completedMessages) {
+		if (group.isComplete() || group.getSequenceSize() == 0) {
+			// The group is complete or else there is no
+			// sequence so there is no more state to track
+			remove(group);
+		} else {
+			// Mark these messages as processed, but do not
+			// remove the group from store
+			if (completedMessages == null) {
+				mark(group);
+			} else {
+				mark(group, completedMessages);
+			}
+		}
+	}
+
 	private final boolean forceComplete(MessageGroup group) {
 
 		Object correlationKey = group.getGroupId();
@@ -232,6 +239,13 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		messageStore.markMessageGroup(group);
 	}
 
+	private void mark(MessageGroup group, Collection<Message> partialSequence) {
+		Object id = group.getGroupId();
+		for (Message message : partialSequence) {
+			messageStore.markMessageFromGroup(id, message);
+		}
+	}
+
 	private void remove(MessageGroup group) {
 		Object correlationKey = group.getGroupId();
 		messageStore.removeMessageGroup(correlationKey);
@@ -271,13 +285,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		completeGroup(first, correlationKey, group);
 	}
 
-	private void completeGroup(Message<?> message, Object correlationKey, MessageGroup group) {
+	private Collection<Message> completeGroup(Message<?> message, Object correlationKey, MessageGroup group) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Completing group with correlationKey ["
 					+ correlationKey + "]");
 		}
 		Object result = outputProcessor.processMessageGroup(group);
+		Collection<Message> partialSequence = null;
+		if (result instanceof Collection<?>) {
+			//Taking a risk here because of Type Erasure. This is covered in the processor contract
+			partialSequence = (Collection<Message>) result;
+		}
 		this.sendReplies(result, message);
+		return partialSequence;
 	}
 
 	private void sendReplies(Object processorResult, Message message) {

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java

@@ -17,7 +17,7 @@ import org.springframework.integration.store.MessageGroup;
 
 /**
  * A processor for <i>correlated</i> groups of messages.
- * 
+ *
  * @author Iwein Fuld
  * @see org.springframework.integration.aggregator.CorrelatingMessageHandler
  */
@@ -27,6 +27,8 @@ public interface MessageGroupProcessor {
 	 * Process the given MessageGroup. Implementations are free to return as few or as many messages based on the
 	 * invocation as needed. For example an aggregating processor will return only a single message representing the
 	 * group, while a resequencing processor will return all messages whose preceding sequence has been satisfied.
+	 * <p/>
+	 * If a multiple messages are returned the return value must be a Collection&lt;Message&gt;.
 	 */
 	Object processMessageGroup(MessageGroup group);
 

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java

@@ -13,18 +13,14 @@
 
 package org.springframework.integration.aggregator;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.store.MessageGroup;
 
+import java.util.*;
+
 /**
  * This class implements all the strategy interfaces needed for a default resequencer.
- * 
+ *
  * @author Iwein Fuld
  * @author Dave Syer
  * @since 2.0
@@ -35,6 +31,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
 
 	/**
 	 * A comparator to use to order messages before processing. The default is to order by sequence number.
+	 *
 	 * @param comparator the comparator to use to order messages
 	 */
 	public void setComparator(Comparator<Message<?>> comparator) {
@@ -43,12 +40,28 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
 
 	public Object processMessageGroup(MessageGroup group) {
 		Collection<Message<?>> messages = group.getUnmarked();
+
 		if (messages.size() > 0) {
 			List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
 			Collections.sort(sorted, this.comparator);
-			return sorted;
+			ArrayList<Message> partialSequence = new ArrayList<Message>();
+			int previousSequence = extractSequenceNumber(sorted.get(0));
+			int currentSequence = previousSequence;
+			for (Message<?> message : sorted) {
+				previousSequence = currentSequence;
+				currentSequence = extractSequenceNumber(message);
+				if (currentSequence - 1 > previousSequence) {
+					//there is a gap in the sequence here
+					break;
+				}
+				partialSequence.add(message);
+			}
+			return partialSequence;
 		}
 		return null;
 	}
 
+	private Integer extractSequenceNumber(Message<?> message) {
+		return message.getHeaders().getSequenceNumber();
+	}
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java

@@ -16,14 +16,14 @@
 
 package org.springframework.integration.aggregator;
 
+import org.springframework.integration.Message;
+import org.springframework.integration.store.MessageGroup;
+
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 
-import org.springframework.integration.Message;
-import org.springframework.integration.store.MessageGroup;
-
 /**
  * An implementation of {@link ReleaseStrategy} that simply compares the
  * current size of the message list to the expected 'sequenceSize'.
@@ -60,9 +60,8 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
 		if (releasePartialSequences) {
 			List<Message<?>> sorted = new ArrayList<Message<?>>(messages.getUnmarked());
 			Collections.sort(sorted, comparator);
-			int head = sorted.get(sorted.size() - 1).getHeaders().getSequenceNumber();
 			int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
-			return tail == messages.getMarked().size() && head - tail == sorted.size();
+			return tail == messages.getMarked().size();
 		}
 		return messages.isComplete();
 	}

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java

@@ -59,4 +59,8 @@ public interface MessageGroup {
 	 */
 	long getTimestamp();
 
+	/**
+	 * Mark the given message in this group. If the message is not part of this group then this call has no effect.
+	 */
+	void mark(Message<?> messageToMark);
 }

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java

@@ -48,13 +48,21 @@ public interface MessageGroupStore {
 	 */
 	MessageGroup markMessageGroup(MessageGroup group);
 
+	/**
+	 * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is no
+	 * longer present in the group.
+	 * @param key the groupId for the group containing the message
+	 * @param messageToRemove the message to be removed
+	 */
+	MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove);
+
 	/**
 	 * Persist a mark on a single message from the group. The group is modified to reflect that 'messageToMark' is no
 	 * longer unmarked but became marked instead.
 	 * @param key the groupId for the group containing the message
 	 * @param messageToMark the message to be marked
 	 */
-	MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark);
+	MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark);
 
 	/**
 	 * Remove the message group with this id.

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java

@@ -13,13 +13,13 @@
 
 package org.springframework.integration.store;
 
+import org.springframework.integration.Message;
+
 import java.util.Collection;
 import java.util.Collections;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 
-import org.springframework.integration.Message;
-
 /**
  * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. The
  * group will grow during its lifetime, when messages are <code>add</code>ed to it. This MessageGroup is thread safe.
@@ -154,6 +154,9 @@ public class SimpleMessageGroup implements MessageGroup {
 		return getOne().getHeaders().getSequenceSize();
 	}
 
+	/**
+	 * {@inheritDoc}}
+	 */
 	public void mark(Message<?> messageToMark) {
 		synchronized (lock) {
 			unmarked.remove(messageToMark);

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -13,17 +13,17 @@
 
 package org.springframework.integration.store;
 
+import org.springframework.integration.Message;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.util.UpperBound;
+import org.springframework.util.Assert;
+
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
-import org.springframework.integration.Message;
-import org.springframework.integration.MessagingException;
-import org.springframework.integration.util.UpperBound;
-import org.springframework.util.Assert;
-
 /**
  * Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the
  * store.
@@ -125,9 +125,15 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
 		groupIdToMessageGroup.remove(groupId);
 	}
 
-	public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark) {
+	public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove) {
 		SimpleMessageGroup group = getMessageGroupInternal(key);
-		group.remove(messageToMark);
+		group.remove(messageToRemove);
+		return group;
+	}
+
+	public MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark) {
+		SimpleMessageGroup group = getMessageGroupInternal(key);
+		group.mark(messageToMark);
 		return group;
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java

@@ -16,15 +16,6 @@
 
 package org.springframework.integration.aggregator;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.integration.Message;
@@ -34,10 +25,19 @@ import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+
+import static org.junit.Assert.*;
+import static org.hamcrest.Matchers.*;
+
 /**
  * @author Marius Bogoevici
  * @author Alex Peters
  * @author Dave Syer
+ * @author Iwein Fuld
  */
 public class ResequencerTests {
 
@@ -65,11 +65,11 @@ public class ResequencerTests {
 		Message<?> reply2 = replyChannel.receive(0);
 		Message<?> reply3 = replyChannel.receive(0);
 		assertNotNull(reply1);
-		assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber());
+		assertThat( reply1.getHeaders().getSequenceNumber(), is(1));
 		assertNotNull(reply2);
-		assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
+		assertThat(reply2.getHeaders().getSequenceNumber(), is(2));
 		assertNotNull(reply3);
-		assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
+		assertThat( reply3.getHeaders().getSequenceNumber(), is(3));
 	}
 
 	@Test
@@ -141,7 +141,8 @@ public class ResequencerTests {
 		assertNull(reply3);
 		// when sending the last message, the whole sequence must have been sent
 		this.resequencer.handleMessage(message4);
-		reply3 = replyChannel.receive(0);		Message<?> reply4 = replyChannel.receive(0);
+		reply3 = replyChannel.receive(0);
+		Message<?> reply4 = replyChannel.receive(0);
 		assertNotNull(reply3);
 		assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
 		assertNotNull(reply4);

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessorTest.java

@@ -0,0 +1,51 @@
+package org.springframework.integration.aggregator;
+
+import org.junit.Test;
+import org.springframework.integration.Message;
+import org.springframework.integration.store.SimpleMessageGroup;
+import org.springframework.integration.support.MessageBuilder;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.assertThat;
+
+/**
+ * @author Iwein Fuld
+ */
+public class ResequencingMessageGroupProcessorTest {
+
+	private ResequencingMessageGroupProcessor processor = new ResequencingMessageGroupProcessor();
+
+	@Test
+	public void shouldProcessSequence() {
+		Message prototypeMessage = MessageBuilder.withPayload("foo").setCorrelationId("x").setSequenceNumber(1).setSequenceSize(3).build();
+		List<Message<?>> unmarked= new ArrayList<Message<?>>();
+		Message message1 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(1).build();
+		Message message2 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(2).build();
+		Message message3 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(3).build();
+		unmarked.add(message1);
+		unmarked.add(message2);
+		unmarked.add(message3);
+		SimpleMessageGroup group = new SimpleMessageGroup(unmarked,"x");
+		List<Message> processedMessages = (List<Message>) processor.processMessageGroup(group);
+		assertThat(processedMessages, hasItems(message1, message2, message3));
+	}
+	
+	@Test
+	public void shouldPartiallProcessIncompleteSequence() {
+		Message prototypeMessage = MessageBuilder.withPayload("foo").setCorrelationId("x").setSequenceNumber(1).setSequenceSize(4).build();
+		List<Message<?>> unmarked= new ArrayList<Message<?>>();
+		Message message2 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(4).build();
+		Message message1 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(1).build();
+		Message message3 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(3).build();
+		unmarked.add(message1);
+		unmarked.add(message2);
+		unmarked.add(message3);
+		SimpleMessageGroup group = new SimpleMessageGroup(unmarked,"x");
+		List<Message> processedMessages = (List<Message>) processor.processMessageGroup(group);
+		assertThat(processedMessages, hasItems(message1));
+		assertThat(processedMessages.size(), is(1));
+	}
+}

spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java

@@ -86,7 +86,9 @@ public class SequenceSizeReleaseStrategyTests {
 		SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
 		releaseStrategy.setReleasePartialSequences(true);
 
-		assertTrue(releaseStrategy.canRelease(groupWithLastAndFirstMessagesOfIncompleteSequence()));
+		boolean canRelease = releaseStrategy.canRelease(groupWithLastAndFirstMessagesOfIncompleteSequence());
+
+		assertTrue(canRelease);
 	}
 
 	private MessageGroup groupWithLastAndFirstMessagesOfIncompleteSequence() {

spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java

@@ -13,33 +13,23 @@
 
 package org.springframework.integration.config;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
-
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.List;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
-import org.springframework.integration.aggregator.CorrelatingMessageHandler;
-import org.springframework.integration.aggregator.CorrelationStrategy;
-import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
-import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
-import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
+import org.springframework.integration.aggregator.*;
 import org.springframework.integration.channel.NullChannel;
-import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 
+import java.util.Comparator;
+
+import static org.junit.Assert.*;
+import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
+
 /**
  * @author Marius Bogoevici
  * @author Mark Fisher
@@ -54,28 +44,6 @@ public class ResequencerParserTests {
 		this.context = new ClassPathXmlApplicationContext("resequencerParserTests.xml", this.getClass());
 	}
 
-	@Test
-	public void testResequencing() {
-		MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel");
-		PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel");
-		List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
-		outboundMessages.add(createMessage("123", "id1", 3, 3, outputChannel));
-		outboundMessages.add(createMessage("789", "id1", 3, 1, outputChannel));
-		outboundMessages.add(createMessage("456", "id1", 3, 2, outputChannel));
-		for (Message<?> message : outboundMessages) {
-			inputChannel.send(message);
-		}
-		Message<?> message1 = outputChannel.receive(500);
-		Message<?> message2 = outputChannel.receive(500);
-		Message<?> message3 = outputChannel.receive(500);
-		assertNotNull(message1);
-		assertEquals(new Integer(1), message1.getHeaders().getSequenceNumber());
-		assertNotNull(message2);
-		assertEquals(new Integer(2), message2.getHeaders().getSequenceNumber());
-		assertNotNull(message3);
-		assertEquals(new Integer(3), message3.getHeaders().getSequenceNumber());
-	}
-
 	@Test
 	public void testDefaultResequencerProperties() {
 		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("defaultResequencer");

spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java

@@ -16,20 +16,15 @@
 
 package org.springframework.integration.store;
 
-import static org.junit.Assert.assertEquals;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-
 import org.junit.Test;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.util.*;
+
+import static org.junit.Assert.assertEquals;
+
 /**
  * @author Dave Syer
  */
@@ -86,7 +81,11 @@ public class MessageStoreTests {
 			throw new UnsupportedOperationException();
 		}
 
-		public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToMark) {
+		public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove) {
+			throw new UnsupportedOperationException();
+		}
+
+		public MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark) {
 			throw new UnsupportedOperationException();
 		}
 

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java

@@ -13,17 +13,6 @@
 
 package org.springframework.integration.jdbc;
 
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Timestamp;
-import java.sql.Types;
-import java.util.Iterator;
-import java.util.List;
-import java.util.UUID;
-
-import javax.sql.DataSource;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.commons.serializer.DeserializingConverter;
@@ -38,16 +27,18 @@ import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageGroup;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.util.UUIDConverter;
-import org.springframework.jdbc.core.JdbcOperations;
-import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.core.PreparedStatementSetter;
-import org.springframework.jdbc.core.RowMapper;
-import org.springframework.jdbc.core.SingleColumnRowMapper;
+import org.springframework.jdbc.core.*;
 import org.springframework.jdbc.support.lob.DefaultLobHandler;
 import org.springframework.jdbc.support.lob.LobHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
+import javax.sql.DataSource;
+import java.sql.*;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+
 /**
  * Implementation of {@link MessageStore} using a relational database via JDBC. SQL scripts to create the necessary
  * tables are packaged as <code>org/springframework/integration/jdbc/schema-*.sql</code>, where <code>*</code> is the
@@ -330,11 +321,29 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 	}
 
-	public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToMark) {
+	public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
+		final String groupKey = getKey(groupId);
+		final String messageId = getKey(messageToRemove.getHeaders().getId());
+
+		jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() {
+			public void setValues(PreparedStatement ps) throws SQLException {
+				logger.debug("Removing message from group with group key=" + groupKey);
+				ps.setString(1, groupKey);
+				ps.setString(2, region);
+				ps.setString(3, messageId);
+			}
+		});
+		return getMessageGroup(groupId);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	public MessageGroup markMessageFromGroup(Object groupId, Message<?> messageToMark) {
 		final String groupKey = getKey(groupId);
 		final String messageId = getKey(messageToMark.getHeaders().getId());
 
-		jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() {
+		jdbcTemplate.update(getQuery(MARK_MESSAGES_IN_GROUP), new PreparedStatementSetter() {
 			public void setValues(PreparedStatement ps) throws SQLException {
 				logger.debug("Removing message from group with group key=" + groupKey);
 				ps.setString(1, groupKey);