INT-4466: Do not force release groups if no match

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

When we schedule group for force complete in the
`AbstractCorrelatingMessageHandler`, we don't track a group `timestamp`
and its `lastModified` before the scheduled task.
This way, in the cluster environment, we may schedule several tasks
for different messages and the first started may release the group too
early.
Just because we extract a `MessageGroup` from the store already in the
task per se.

* Propagate the actual `timestamp` and `lastModified` from group before
scheduling task.
Compare these value with the actual group metadata in the
`processForceRelease()` before performing real `forceRelease()`.
This way we restore behavior before fixing memory leak, when we
propagated full `MessageGroup` to the scheduled task
* Implement `ResequencingMessageHandler.getComponentType()` for
consistency
* Remove unnecessary overhead with the `volatile` on many
`AbstractCorrelatingMessageHandler` properties, which hardly ever can be
changed at runtime

**Cherry-pick to 5.0.x and 4.3.x**

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -96,39 +96,39 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private final MessageGroupProcessor outputProcessor;
 
-	private volatile MessageGroupStore messageStore;
+	private MessageGroupStore messageStore;
 
-	private volatile CorrelationStrategy correlationStrategy;
+	private CorrelationStrategy correlationStrategy;
 
-	private volatile ReleaseStrategy releaseStrategy;
+	private ReleaseStrategy releaseStrategy;
 
-	private volatile MessageChannel discardChannel;
+	private MessageChannel discardChannel;
 
-	private volatile String discardChannelName;
+	private String discardChannelName;
 
 	private boolean sendPartialResultOnExpiry = false;
 
-	private volatile boolean sequenceAware = false;
+	private boolean sequenceAware = false;
 
-	private volatile LockRegistry lockRegistry = new DefaultLockRegistry();
+	private LockRegistry lockRegistry = new DefaultLockRegistry();
 
 	private boolean lockRegistrySet = false;
 
-	private volatile long minimumTimeoutForEmptyGroups;
+	private long minimumTimeoutForEmptyGroups;
 
-	private volatile boolean releasePartialSequences;
+	private boolean releasePartialSequences;
 
-	private volatile Expression groupTimeoutExpression;
+	private Expression groupTimeoutExpression;
 
-	private volatile List<Advice> forceReleaseAdviceChain;
+	private List<Advice> forceReleaseAdviceChain;
 
 	private MessageGroupProcessor forceReleaseProcessor = new ForceReleaseMessageGroupProcessor();
 
 	private EvaluationContext evaluationContext;
 
-	private volatile ApplicationEventPublisher applicationEventPublisher;
+	private ApplicationEventPublisher applicationEventPublisher;
 
-	private volatile boolean expireGroupsUponTimeout = true;
+	private boolean expireGroupsUponTimeout = true;
 
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
 			CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
@@ -415,7 +415,7 @@ protected void handleMessageInternal(Message<?> message) throws Exception {
 				if (this.releaseStrategy.canRelease(messageGroup)) {
 					Collection<Message<?>> completedMessages = null;
 					try {
-						completedMessages = this.completeGroup(message, correlationKey, messageGroup);
+						completedMessages = completeGroup(message, correlationKey, messageGroup);
 					}
 					finally {
 						// Always clean up even if there was an exception
@@ -445,13 +445,15 @@ private void scheduleGroupToForceComplete(MessageGroup messageGroup) {
 		if (groupTimeout != null && groupTimeout >= 0) {
 			if (groupTimeout > 0) {
 				final Object groupId = messageGroup.getGroupId();
+				final long timestamp = messageGroup.getTimestamp();
+				final long lastModified = messageGroup.getLastModified();
 				ScheduledFuture<?> scheduledFuture = getTaskScheduler()
 						.schedule(new Runnable() {
 
 							@Override
 							public void run() {
 								try {
-									processForceRelease(groupId);
+									processForceRelease(groupId, timestamp, lastModified);
 								}
 								catch (MessageDeliveryException e) {
 									if (logger.isWarnEnabled()) {
@@ -480,9 +482,11 @@ private void scheduleGroupToForceComplete(Object groupId) {
 		scheduleGroupToForceComplete(messageGroup);
 	}
 
-	private void processForceRelease(Object groupId) {
+	private void processForceRelease(Object groupId, long timestamp, long lastModified) {
 		MessageGroup messageGroup = this.messageStore.getMessageGroup(groupId);
-		this.forceReleaseProcessor.processMessageGroup(messageGroup);
+		if (messageGroup.getTimestamp() == timestamp && messageGroup.getLastModified() == lastModified) {
+			this.forceReleaseProcessor.processMessageGroup(messageGroup);
+		}
 	}
 
 	private void discardMessage(Message<?> message) {
@@ -546,9 +550,11 @@ protected void forceComplete(MessageGroup group) {
 				}
 				long lastModifiedNow = groupNow.getLastModified();
 				int groupSize = groupNow.size();
+
 				if ((!groupNow.isComplete() || groupSize == 0)
 						&& group.getLastModified() == lastModifiedNow
 						&& group.getTimestamp() == groupNow.getTimestamp()) {
+
 					if (groupSize > 0) {
 						if (this.releaseStrategy.canRelease(groupNow)) {
 							completeGroup(correlationKey, groupNow);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -65,6 +65,11 @@ public final void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) {
 		super.setExpireGroupsUponTimeout(expireGroupsUponTimeout);
 	}
 
+	@Override
+	public String getComponentType() {
+		return "resequencer";
+	}
+
 	@Override
 	protected boolean shouldCopyRequestHeaders() {
 		return false;