INT-3387: MessageGroupStore Improvements

JIRA: https://jira.spring.io/browse/INT-3387,
https://jira.spring.io/browse/INT-3806

* Introduce
```
MessageGroupStore

void addMessagesToGroup(Object groupId, Message<?>... messages);
```
And implement it in all stores.

* Use new `addMessagesToGroup` where it is reasonable, e.g. `DelayHandler`
* Optimize test-case to use a new store method (where it is possible)
* Fix timing delays in the `JdbcMessageStoreTests`
* Introduce `PersistentMessageGroup`
* Add `AbstractMessageGroupStore#proxyMessageGroupForLazyLoad` to wrap the raw `MessageGroup` to the `PersistentMessageGroup` for lazy-load
* Rework `MessageGroupMetadata` do not be `immutable` and allow to store/restore in the `AbstractKeyValueMessageStore` only the `MessageGroupMetadata`
* Refactor `ResequencingMessageHandler` and `SequenceSizeReleaseStrategy` a bit for better performance when interact with the `MessageGroup`
* Add `AbstractMessageGroupStore#setLazyLoadMessageGroups` to switch off the `lazy-load` behavior and restore the previous full `MessageGroup` logic
* Add `What's New` note and `message-store.adoc` paragraph for the lazy-load functionality

`GroupType.PERSISTENT` and not lazy by default

PR Comments

Fix `JdbcMessageStoreTests` timing issues

Address PR comments

* Add performance test to the `ConfigurableMongoDbMessageGroupStoreTests`
* Add JavaDocs for the `MessageGroupFactory` methods
* Add `log4j.properties` into the `test` MongoDB module for better traceability
* Fix `JdbcMessageStore#getOneMessageFromGroup()` over the `doPollForMessage()` delegation.
The `jdbcTemplate.queryForObject()` requires exactly one and only one raw in `resultSet`
* Add performance test results into the `message-store.adoc`

Author: artembilan

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

@@ -95,7 +95,7 @@ protected void handleMessageInternal(Message<?> message) throws Exception {
 		Object correlationKey = this.correlationStrategy.getCorrelationKey(message);
 		Object lock = getLock(correlationKey);
 		synchronized (lock) {
-			this.store.addMessageToGroup(correlationKey, message);
+			this.store.addMessagesToGroup(correlationKey, message);
 		}
 		if (log.isDebugEnabled()) {
 			log.debug(String.format("Handled message for key [%s]: %s.", correlationKey, message));

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

@@ -18,7 +18,6 @@
 
 import java.util.Collection;
 
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
@@ -74,12 +73,9 @@ protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> co
 
 	@Override
 	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages, boolean timeout) {
-		int size = messageGroup.getMessages().size();
-		int sequenceSize = 0;
-		Message<?> message = messageGroup.getOne();
-		if (message != null) {
-			sequenceSize = new IntegrationMessageHeaderAccessor(message).getSequenceSize();
-		}
+		int size = messageGroup.size();
+		int sequenceSize = messageGroup.getSequenceSize();
+
 		// If there is no sequence then it must be incomplete or unbounded
 		if (sequenceSize > 0 && sequenceSize == size) {
 			remove(messageGroup);

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

@@ -36,6 +36,7 @@
  * @author Dave Syer
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
  * @author Enrique Rodríguez
  */
 public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
@@ -55,9 +56,9 @@ public SequenceSizeReleaseStrategy(boolean releasePartialSequences) {
 	}
 
 	/**
-	 * Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be
-	 * ordered they will be released, provided they all have sequence numbers greater than those already released.
-	 *
+	 * Flag that determines if partial sequences are allowed. If true then as soon as
+	 * enough messages arrive that can be ordered they will be released, provided they
+	 * all have sequence numbers greater than those already released.
 	 * @param releasePartialSequences true when partial sequences should be released.
 	 */
 	public void setReleasePartialSequences(boolean releasePartialSequences) {
@@ -69,13 +70,12 @@ public boolean canRelease(MessageGroup messageGroup) {
 
 		boolean canRelease = false;
 
-		Collection<Message<?>> messages = messageGroup.getMessages();
-
-		if (this.releasePartialSequences && !messages.isEmpty()) {
-
+		int size = messageGroup.size();
+		if (this.releasePartialSequences && size > 0) {
 			if (logger.isTraceEnabled()) {
 				logger.trace("Considering partial release of group [" + messageGroup + "]");
 			}
+			Collection<Message<?>> messages = messageGroup.getMessages();
 			Message<?> minMessage = Collections.min(messages, this.comparator);
 
 			int nextSequenceNumber = new IntegrationMessageHeaderAccessor(minMessage).getSequenceNumber();
@@ -86,13 +86,11 @@ public boolean canRelease(MessageGroup messageGroup) {
 			}
 		}
 		else {
-			int size = messages.size();
-
 			if (size == 0) {
 				canRelease = true;
 			}
 			else {
-				int sequenceSize = new IntegrationMessageHeaderAccessor(messageGroup.getOne()).getSequenceSize();
+				int sequenceSize = messageGroup.getSequenceSize();
 				// If there is no sequence then it must be incomplete....
 				if (sequenceSize == size) {
 					canRelease = true;

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

@@ -94,38 +94,62 @@ public long getMessageCount() {
 	 */
 	@Override
 	public MessageGroup getMessageGroup(Object groupId) {
-		return buildMessageGroup(groupId, false);
-	}
+		MessageGroupMetadata metadata = getGroupMetadata(groupId);
+		if (metadata != null) {
 
+			MessageGroup messageGroup = getMessageGroupFactory()
+					.create(this, groupId, metadata.getTimestamp(), metadata.isComplete());
+			messageGroup.setLastModified(metadata.getLastModified());
+			messageGroup.setLastReleasedMessageSequenceNumber(metadata.getLastReleasedMessageSequenceNumber());
+			return messageGroup;
+		}
+		else {
+			return new SimpleMessageGroup(groupId);
+		}
+	}
 
-	/**
-	 * Add a Message to the group with the provided group ID.
-	 */
 	@Override
-	public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
+	public MessageGroupMetadata getGroupMetadata(Object groupId) {
 		Assert.notNull(groupId, "'groupId' must not be null");
-		Assert.notNull(message, "'message' must not be null");
+		Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
+		if (mgm != null) {
+			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
+			return (MessageGroupMetadata) mgm;
+		}
+		return null;
+	}
 
-		// add message as is to the MG accessible by the caller
-		MessageGroup messageGroup = getMessageGroup(groupId);
+	@Override
+	public void addMessagesToGroup(Object groupId, Message<?>... messages) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		Assert.notNull(messages, "'messages' must not be null");
 
-		messageGroup.add(message);
+		MessageGroupMetadata metadata = getGroupMetadata(groupId);
+		SimpleMessageGroup group = null;
+		if (metadata == null) {
+			group = new SimpleMessageGroup(groupId);
+		}
 
-		// enrich Message with additional headers and add it to MS
-		Message<?> enrichedMessage = enrichMessage(message);
+		for (Message<?> message : messages) {
+			// enrich Message with additional headers and add it to MS
+			Message<?> enrichedMessage = enrichMessage(message);
+			addMessage(enrichedMessage);
+			if (metadata != null) {
+				metadata.add(enrichedMessage.getHeaders().getId());
+			}
+			else {
+				group.add(enrichedMessage);
+			}
+		}
 
-		addMessage(enrichedMessage);
+		if (group != null) {
+			metadata = new MessageGroupMetadata(group);
+		}
 
-		// build raw MessageGroup and add enriched Message to it
-		MessageGroup rawGroup = buildMessageGroup(groupId, true);
-		rawGroup.setLastModified(System.currentTimeMillis());
-		rawGroup.add(enrichedMessage);
+		metadata.setLastModified(System.currentTimeMillis());
 
 		// store MessageGroupMetadata built from enriched MG
-		doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
-
-		// return clean MG
-		return getMessageGroup(groupId);
+		doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata);
 	}
 
 	/**
@@ -137,32 +161,17 @@ public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToR
 		Assert.notNull(groupId, "'groupId' must not be null");
 		Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
 
-		// build raw MG
-		MessageGroup rawGroup = buildMessageGroup(groupId, true);
-
-		// create a clean instance of
-		MessageGroup messageGroup = normalizeSimpleMessageGroup(rawGroup);
-
-
-		Message<?> actualMessageToRemove = null;
-
-		for (Message<?> message : rawGroup.getMessages()) {
-			if (message.getHeaders().getId().equals(messageToRemove.getHeaders().getId())) {
-				actualMessageToRemove = message;
-				break;
-			}
-		}
-
-		if (actualMessageToRemove != null) {
-			rawGroup.remove(actualMessageToRemove);
-			removeMessage(messageToRemove.getHeaders().getId());
-			rawGroup.setLastModified(System.currentTimeMillis());
+		UUID id = messageToRemove.getHeaders().getId();
+		removeMessage(id);
 
-			doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
-			messageGroup = getMessageGroup(groupId);
+		MessageGroupMetadata metadata = getGroupMetadata(groupId);
+		if (metadata != null) {
+			metadata.remove(id);
+			metadata.setLastModified(System.currentTimeMillis());
+			doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata);
 		}
 
-		return messageGroup;
+		return getMessageGroup(groupId);
 	}
 
 
@@ -188,10 +197,12 @@ public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messa
 	@Override
 	public void completeGroup(Object groupId) {
 		Assert.notNull(groupId, "'groupId' must not be null");
-		MessageGroup messageGroup = buildMessageGroup(groupId, true);
-		messageGroup.complete();
-		messageGroup.setLastModified(System.currentTimeMillis());
-		doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+		MessageGroupMetadata metadata = getGroupMetadata(groupId);
+		if (metadata != null) {
+			metadata.complete();
+			metadata.setLastModified(System.currentTimeMillis());
+			doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata);
+		}
 	}
 
 	/**
@@ -215,37 +226,62 @@ public void removeMessageGroup(Object groupId) {
 	@Override
 	public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
 		Assert.notNull(groupId, "'groupId' must not be null");
-		MessageGroup messageGroup = buildMessageGroup(groupId, true);
-		messageGroup.setLastReleasedMessageSequenceNumber(sequenceNumber);
-		messageGroup.setLastModified(System.currentTimeMillis());
-		doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+		MessageGroupMetadata metadata = getGroupMetadata(groupId);
+		if (metadata == null) {
+			SimpleMessageGroup messageGroup = new SimpleMessageGroup(groupId);
+			metadata = new MessageGroupMetadata(messageGroup);
+		}
+		metadata.setLastReleasedMessageSequenceNumber(sequenceNumber);
+		metadata.setLastModified(System.currentTimeMillis());
+		doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata);
 	}
 
 	@Override
 	public Message<?> pollMessageFromGroup(Object groupId) {
-		Assert.notNull(groupId, "'groupId' must not be null");
-		Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
-		if (mgm != null) {
-			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
-			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
-
-			UUID firstId = messageGroupMetadata.firstId();
+		MessageGroupMetadata groupMetadata = getGroupMetadata(groupId);
+		if (groupMetadata != null) {
+			UUID firstId = groupMetadata.firstId();
 			if (firstId != null) {
-				messageGroupMetadata.remove(firstId);
-				messageGroupMetadata.setLastModified(System.currentTimeMillis());
-				doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata);
+				groupMetadata.remove(firstId);
+				groupMetadata.setLastModified(System.currentTimeMillis());
+				doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, groupMetadata);
 				return removeMessage(firstId);
 			}
 		}
 		return null;
 	}
 
+	@Override
+	public Message<?> getOneMessageFromGroup(Object groupId) {
+		MessageGroupMetadata groupMetadata = getGroupMetadata(groupId);
+		if (groupMetadata != null) {
+			UUID messageId = groupMetadata.firstId();
+			if (messageId != null) {
+				return getMessage(messageId);
+			}
+		}
+		return null;
+	}
+
+	@Override
+	public Collection<Message<?>> getMessagesForGroup(Object groupId) {
+		MessageGroupMetadata groupMetadata = getGroupMetadata(groupId);
+		ArrayList<Message<?>> messages = new ArrayList<Message<?>>();
+		if (groupMetadata != null) {
+			Iterator<UUID> messageIds = groupMetadata.messageIdIterator();
+			while (messageIds.hasNext()) {
+				messages.add(getMessage(messageIds.next()));
+			}
+		}
+		return messages;
+	}
+
 	@Override
 	@SuppressWarnings("unchecked")
 	public Iterator<MessageGroup> iterator() {
 		final Iterator<?> idIterator = normalizeKeys(
 				(Collection<String>) doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*"))
-					.iterator();
+				.iterator();
 		return new MessageGroupIterator(idIterator);
 	}
 
@@ -266,13 +302,13 @@ else if (strKey.startsWith(MESSAGE_KEY_PREFIX)) {
 
 	@Override
 	public int messageGroupSize(Object groupId) {
-		Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
+		MessageGroupMetadata mgm = getGroupMetadata(groupId);
 		if (mgm != null) {
-			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
-			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
-			return messageGroupMetadata.size();
+			return mgm.size();
+		}
+		else {
+			return 0;
 		}
-		return 0;
 	}
 
 	protected abstract Object doRetrieve(Object id);
@@ -308,48 +344,6 @@ private Message<?> enrichMessage(Message<?> message) {
 		return enrichedMessage;
 	}
 
-	private MessageGroup buildMessageGroup(Object groupId, boolean raw) {
-		Assert.notNull(groupId, "'groupId' must not be null");
-		Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
-		if (mgm != null) {
-			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
-			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
-			ArrayList<Message<?>> messages = new ArrayList<Message<?>>();
-
-			Iterator<UUID> messageIds = messageGroupMetadata.messageIdIterator();
-			while (messageIds.hasNext()) {
-				UUID next = messageIds.next();
-				if (next != null) {
-					if (raw) {
-						messages.add(getRawMessage(next));
-					}
-					else {
-						messages.add(getMessage(next));
-					}
-				}
-			}
-
-			MessageGroup messageGroup = getMessageGroupFactory()
-					.create(messages, groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete());
-			messageGroup.setLastModified(messageGroupMetadata.getLastModified());
-			messageGroup.setLastReleasedMessageSequenceNumber(
-					messageGroupMetadata.getLastReleasedMessageSequenceNumber());
-			return messageGroup;
-		}
-		else {
-			return getMessageGroupFactory().create(groupId);
-		}
-	}
-
-	private MessageGroup normalizeSimpleMessageGroup(MessageGroup messageGroup) {
-		MessageGroup normalizedGroup = getMessageGroupFactory().create(messageGroup.getGroupId());
-		for (Message<?> message : messageGroup.getMessages()) {
-			Message<?> normalizedMessage = normalizeMessage(message);
-			normalizedGroup.add(normalizedMessage);
-		}
-		return normalizedGroup;
-	}
-
 	private Message<?> getRawMessage(UUID id) {
 		Assert.notNull(id, "'id' must not be null");
 		Object message = doRetrieve(MESSAGE_KEY_PREFIX + id);
@@ -380,4 +374,5 @@ public void remove() {
 			throw new UnsupportedOperationException();
 		}
 	}
+
 }