INT-3876: Track groups in correlation endpoints

JIRA: https://jira.spring.io/browse/INT-3876

INT-3876 track groups ids in AbstractCorrelatingMessageHandler and docs

INT-3876 Fix merge conflict and add missing link in doc

INT-3876 Move groupId check in message group processor and rework on docs

INT-3876 Remove extra space from the doc

INT-3876 Fix checkstyle refer instance variable groupid correctly

INT-3876 Update aggregator doc with more details

* Polishing `aggregator.adoc`
* Rework `AbstractCorrelatingMessageHandlerTests.testDontReapMessageOfOtherHandler`
do not use redundant options

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

@@ -21,8 +21,10 @@ import java.util.Collections;
 import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.locks.Lock;
@@ -75,6 +77,9 @@ import org.springframework.util.CollectionUtils;
  * By default the {@link CorrelationStrategy} will be a
  * {@link HeaderAttributeCorrelationStrategy} and the {@link ReleaseStrategy} will be a
  * {@link SequenceSizeReleaseStrategy}.
+ * <p>
+ * Use proper {@link CorrelationStrategy} for cases when same {@link MessageStore} is used
+ * for multiple handlers to ensure uniqueness of message groups across handlers.
  *
  * @author Iwein Fuld
  * @author Dave Syer
@@ -83,7 +88,7 @@ import org.springframework.util.CollectionUtils;
  * @author Artem Bilan
  * @author David Liu
  * @author Enrique Rodriguez
- *
+ * @author Meherzad Lahewala
  * @since 2.0
  */
 public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler
@@ -95,6 +100,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<>();
 
+	private final Set<Object> groupIds =  new HashSet<>();
+
 	private MessageGroupProcessor outputProcessor;
 
 	private volatile MessageGroupStore messageStore;
@@ -481,7 +488,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 								if (this.logger.isDebugEnabled()) {
 									this.logger.debug("Removing empty group: " + groupUuid);
 								}
-								this.messageStore.removeMessageGroup(groupId);
+								remove(messageGroup);
 							}
 						}
 						finally {
@@ -671,9 +678,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		}
 	}
 
-	void remove(MessageGroup group) {
+	protected void remove(MessageGroup group) {
 		Object correlationKey = group.getGroupId();
 		this.messageStore.removeMessageGroup(correlationKey);
+		this.groupIds.remove(group.getGroupId());
 	}
 
 	protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence) {
@@ -682,6 +690,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 	}
 
 	protected MessageGroup store(Object correlationKey, Message<?> message) {
+		this.groupIds.add(correlationKey);
 		return this.messageStore.addMessageToGroup(correlationKey, message);
 	}
 
@@ -851,7 +860,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 		@Override
 		public Object processMessageGroup(MessageGroup group) {
-			forceComplete(group);
+			if (AbstractCorrelatingMessageHandler.this.groupIds.contains(group.getGroupId())) {
+				forceComplete(group);
+			}
 			return null;
 		}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -52,6 +52,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Meherzad Lahewala
+ *
  * @since 2.2
  *
  */
@@ -417,4 +419,29 @@ public class AbstractCorrelatingMessageHandlerTests {
 		assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
 	}
 
+	@Test
+	public void testDontReapMessageOfOtherHandler() throws Exception {
+		MessageGroupStore groupStore = new SimpleMessageStore();
+
+		AggregatingMessageHandler handler1 = new AggregatingMessageHandler(group -> group, groupStore);
+		AggregatingMessageHandler handler2 = new AggregatingMessageHandler(group -> group, groupStore);
+
+		QueueChannel handler1DiscardChannel = new QueueChannel();
+		handler1.setDiscardChannel(handler1DiscardChannel);
+
+		QueueChannel handler2DiscardChannel = new QueueChannel();
+		handler2.setDiscardChannel(handler2DiscardChannel);
+
+		handler1.setReleaseStrategy(group -> false);
+		handler2.setReleaseStrategy(group -> false);
+
+		handler1.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
+		handler1.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
+		handler2.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("bar").build());
+
+		groupStore.expireMessageGroups(0);
+
+		assertTrue(handler1DiscardChannel.getQueueSize() == 2);
+		assertTrue(handler2DiscardChannel.getQueueSize() == 1);
+	}
 }

src/reference/asciidoc/aggregator.adoc

@@ -846,8 +846,8 @@ If the flag is set to true, then when the expiry callback is invoked, any unmark
 
 [IMPORTANT]
 =====
-When using a `MessageGroupStoreReaper`, it is generally recommended to use a separate `MessageStore` for each correlating endpoint.
-Otherwise, unexpected results may occur because one endpoint may remove another endpoint's groups.
+When a shared `MessageStore` is used for different correlation endpoints, it is necessary to configure a proper `CorrelationStrategy` to ensure uniqueness for group ids.
+Otherwise unexpected behavior may happen when one correlation endpoint may release or expire messages from others - messages with the same correlation key are stored in the same message group.
 
 Some `MessageStore` implementations allow using the same physical resources, by partitioning the data; for example, the `JdbcMessageStore` has a `region` property; the `MongoDbMessageStore` has a `collectionName` property.