diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
index 4df188f150..e129bc99b7 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
@@ -268,20 +268,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 			lock.lockInterruptibly();
 			try {
 				/*
-				 * Need to verify the group hasn't changed while we were waiting on
-				 * its lock. We have to re-fetch the group for this. A possible
-				 * future improvement would be to add MessageGroupStore.getLastModified(groupId).
+				 * Refetch the group because it might have changed while we were waiting on
+				 * its lock. If the last modified timestamp changed, defer the completion
+				 * because the selection condition may have changed such that the group
+				 * would no longer be eligible.
 				 */
-				MessageGroup messageGroupNow = this.messageStore.getMessageGroup(
+				MessageGroup groupNow = this.messageStore.getMessageGroup(
 						group.getGroupId());
-				long lastModifiedNow = messageGroupNow.getLastModified();
+				long lastModifiedNow = groupNow.getLastModified();
 				if (group.getLastModified() == lastModifiedNow) {
-					if (group.size() > 0) {
-						if (releaseStrategy.canRelease(group)) {
-							this.completeGroup(correlationKey, group);
+					if (groupNow.size() > 0) {
+						if (releaseStrategy.canRelease(groupNow)) {
+							this.completeGroup(correlationKey, groupNow);
 						}
 						else {
-							this.expireGroup(correlationKey, group);
+							this.expireGroup(correlationKey, groupNow);
 						}
 					}
 					else {
@@ -292,14 +293,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 						 */
 						removeGroup = lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
 						if (removeGroup && logger.isDebugEnabled()) {
-							logger.debug("Removing empty group: " + group.getGroupId());
+							logger.debug("Removing empty group: " + correlationKey);
 						}
 					}
 				}
 				else {
 					removeGroup = false;
 					if (logger.isDebugEnabled()) {
-						logger.debug("Group expiry candidate (" + group.getGroupId() +
+						logger.debug("Group expiry candidate (" + correlationKey +
 								") has changed - it may be reconsidered for a future expiration");
 					}
 				}
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
index 6ef72047d1..ecc647d3bd 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
@@ -16,9 +16,11 @@
 package org.springframework.integration.aggregator;
 
 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 java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -28,9 +30,11 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
@@ -249,4 +253,34 @@ public class AbstractCorrelatingMessageHandlerTests {
 		assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
 	}
 
+	@Test
+	public void testReapWithChangeInSameMillisecond() throws Exception {
+		MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
+		handler.setReleaseStrategy(new ReleaseStrategy() {
+			public boolean canRelease(MessageGroup group) {
+				return true;
+			}
+		});
+		QueueChannel outputChannel = new QueueChannel();
+		handler.setOutputChannel(outputChannel);
+		MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
+		Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class);
+		forceComplete.setAccessible(true);
+		mgs.addMessageToGroup("foo", new GenericMessage<String>("foo"));
+		GenericMessage<String> secondMessage = new GenericMessage<String>("bar");
+		mgs.addMessageToGroup("foo", secondMessage);
+		MessageGroup group = mgs.getMessageGroup("foo");
+		// remove a message
+		mgs.removeMessageFromGroup("foo", secondMessage);
+		// force lastModified to be the same
+		MessageGroup groupNow = mgs.getMessageGroup("foo");
+		new DirectFieldAccessor(group).setPropertyValue("lastModified", groupNow.getLastModified());
+		forceComplete.invoke(handler, group);
+		Message<?> message = outputChannel.receive(0);
+		assertNotNull(message);
+		Collection<?> payload = (Collection<?>) message.getPayload();
+		assertEquals(1, payload.size());
+	}
+
 }