INT-4193: Large Group Aggregation Performance

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

When using the internal `SequenceAwareMessageGroup` within an correlating
message handler, the messages were copied to a new collection before
checking for duplicate sequences in `canAdd()`.

This was unnecessary since we never add anything to this group, if `canAdd()`
returns true, the message is added to the store; this group is discarded.

Instead, use the message collection from the original group; although it is
not modifiable, this is not an issue because we don't need to modify it.

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

@@ -693,13 +693,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 	protected static class SequenceAwareMessageGroup extends SimpleMessageGroup {
 
 		public SequenceAwareMessageGroup(MessageGroup messageGroup) {
-			super(messageGroup);
+			/*
+			 * Since this group is temporary, and never added to, we simply use the
+			 * supplied group's message collection for the lookup rather than creating a
+			 * new group.
+			 */
+			super(messageGroup.getMessages(), null, messageGroup.getGroupId(), messageGroup.getTimestamp(),
+					messageGroup.isComplete(), true);
 		}
 
 		/**
-		 * This method determines whether messages have been added to this group that supersede the given message based on
-		 * its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size
-		 * or sequences that are missing certain sequence numbers.
+		 * This method determines whether messages have been added to this group that
+		 * supersede the given message based on its sequence id. This can be helpful to
+		 * avoid ending up with sequences larger than their required sequence size or
+		 * sequences that are missing certain sequence numbers.
 		 */
 		@Override
 		public boolean canAdd(Message<?> message) {

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aggregator;
+
+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'. It does not support releasing partial
+ * sequences. Correlating message handlers using this strategy do not check for duplicate
+ * sequence numbers.
+ * @author Gary Russell
+ * @since 4.3.7
+ *
+ */
+public class SimpleSequenceSizeReleaseStrategy implements ReleaseStrategy {
+
+	@Override
+	public boolean canRelease(MessageGroup group) {
+		return group.getSequenceSize() == group.size();
+	}
+
+}

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

@@ -66,21 +66,23 @@ public class SimpleMessageGroup implements MessageGroup {
 	}
 
 	public SimpleMessageGroup(Collection<? extends Message<?>> messages, Object groupId, long timestamp,
-	                          boolean complete) {
-		this(new LinkedHashSet<Message<?>>(), messages, groupId, timestamp, complete);
+			boolean complete) {
+		this(new LinkedHashSet<Message<?>>(), messages, groupId, timestamp, complete, false);
 	}
 
-	SimpleMessageGroup(Collection<Message<?>> internalStore, Collection<? extends Message<?>> messages, Object groupId,
-	                   long timestamp, boolean complete) {
+	protected SimpleMessageGroup(Collection<Message<?>> internalStore, Collection<? extends Message<?>> messages,
+			Object groupId, long timestamp, boolean complete, boolean storePreLoaded) {
 		Assert.notNull(internalStore, "'internalStore' must not be null");
-		Assert.notNull(messages, "'messages' must not be null");
 		this.messages = internalStore;
 		this.groupId = groupId;
 		this.timestamp = timestamp;
 		this.complete = complete;
-		for (Message<?> message : messages) {
-			if (message != null) { //see INT-2666
-				addMessage(message);
+		if (!storePreLoaded) {
+			Assert.notNull(messages, "'messages' must not be null");
+			for (Message<?> message : messages) {
+				if (message != null) { //see INT-2666
+					addMessage(message);
+				}
 			}
 		}
 	}

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

@@ -56,7 +56,7 @@ public class SimpleMessageGroupFactory implements MessageGroupFactory {
 	@Override
 	public MessageGroup create(Collection<? extends Message<?>> messages, Object groupId, long timestamp,
 							   boolean complete) {
-		return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete);
+		return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete, false);
 	}
 
 	@Override

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

@@ -26,10 +26,12 @@ import java.lang.reflect.Constructor;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 
 import org.junit.Test;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
@@ -40,6 +42,7 @@ import org.springframework.util.StopWatch;
  * @author Oleg Zhurakousky
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Gary Russell
  */
 public class SimpleMessageGroupTests {
 
@@ -54,11 +57,12 @@ public class SimpleMessageGroupTests {
 		Constructor<SimpleMessageGroup> ctr = clazz.getDeclaredConstructor(MessageGroup.class);
 		ctr.setAccessible(true);
 		group = ctr.newInstance(group);
+		new DirectFieldAccessor(group).setPropertyValue("messages", new HashSet<Message<?>>());
 	}
 
 	@Test
 	public void shouldFindSupersedingMessagesIfSequenceAware() throws Exception {
-		this.prepareForSequenceAwareMessageGroup();
+		prepareForSequenceAwareMessageGroup();
 		final Message<?> message1 = MessageBuilder.withPayload("test").setSequenceNumber(1).build();
 		final Message<?> message2 = MessageBuilder.fromMessage(message1).setSequenceNumber(1).build();
 		assertThat(group.canAdd(message1), is(true));
@@ -69,7 +73,7 @@ public class SimpleMessageGroupTests {
 
 	@Test
 	public void shouldIgnoreMessagesWithZeroSequenceNumberIfSequenceAware() throws Exception {
-		this.prepareForSequenceAwareMessageGroup();
+		prepareForSequenceAwareMessageGroup();
 		final Message<?> message1 = MessageBuilder.withPayload("test").build();
 		final Message<?> message2 = MessageBuilder.fromMessage(message1).build();
 		assertThat(group.canAdd(message1), is(true));

src/reference/asciidoc/aggregator.adoc

@@ -218,6 +218,35 @@ Spring Integration provides an out-of-the box implementation for `ReleaseStrateg
 This implementation consults the `SEQUENCE_NUMBER` and `SEQUENCE_SIZE` headers of each arriving message to decide when a message group is complete and ready to be aggregated.
 As shown above, it is also the default strategy.
 
+If you are aggregating large groups, you don't need to release partial groups, and you don't need to detect/reject duplicate sequences, consider using the `SimpleSequenceSizeReleaseStrategy` instead - it is much more efficient for these use cases, and will be the default in future releases when partial group release is not specified.
+
+===== Aggregating Large Groups
+
+The 4.3 release changed the default `Collection` for messages in a `SimpleMessageGroup` to `HashSet` (it was previously a `BlockingQueue`).
+This was expensive when removing individual messages from large groups (an O(n) linear scan was required).
+Although the hash set is generally much faster for removing, it can be expensive for large messages because the hash has to be calculated (on both inserts and removes).
+If you have messages that are expensive to hash, consider using some other collection type.
+As discussed in <<message-group-factory>>, a `SimpleMessageGroupFactory` is provided so you can select the `Collection` that best suits your needs.
+You can also provide your own factory implementation to create some other `Collection<Message<?>>`.
+
+Here is an example of how to configure an aggregator with the previous implementation and a `SimpleSequenceSizeReleaseStrategy`.
+
+[source, xml]
+----
+<int:aggregator input-channel="aggregate"
+    output-channel="out" message-store="store" release-strategy="releaser" />
+
+<bean id="store" class="org.springframework.integration.store.SimpleMessageStore">
+    <property name="messageGroupFactory">
+        <bean class="org.springframework.integration.store.SimpleMessageGroupFactory">
+            <constructor-arg value="BLOCKING_QUEUE"/>
+        </bean>
+    </property>
+</bean>
+
+<bean id="releaser" class="SimpleSequenceSizeReleaseStrategy" />
+----
+
 ===== CorrelationStrategy
 
 The `CorrelationStrategy` interface is defined as follows:
@@ -629,9 +658,9 @@ For this purpose the `groupTimeout` option allows scheduling the `MessageGroup`
 [source,xml]
 ----
 <aggregator input-channel="input" output-channel="output"
-		send-partial-result-on-expiry="true"
-		group-timeout-expression="size() ge 2 ? 10000 : -1"
-		release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
+        send-partial-result-on-expiry="true"
+        group-timeout-expression="size() ge 2 ? 10000 : -1"
+        release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
 ----
 
 With this example, the normal _release_ will be possible if the aggregator receives the last message in sequence as defined by the `release-strategy-expression`.