INT-3529 Polishing

Move group completion to the Resequencer's `afterRelease()` when
the group is timed out.

Author: garyrussell

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

@@ -483,6 +483,17 @@ private void discardMessage(Message<?> message) {
 	 */
 	protected abstract void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages);
 
+	/**
+	 * Subclasses may override if special action is needed because the group was released or discarded
+	 * due to a timeout. By default, {@link #afterRelease(MessageGroup, Collection)} is invoked.
+	 * @param group The group.
+	 * @param completedMessages The completed messages.
+	 * @param timeout True if the release/discard was due to a timeout.
+	 */
+	protected void  afterRelease(MessageGroup group, Collection<Message<?>> completedMessages, boolean timeout) {
+		afterRelease(group, completedMessages);
+	}
+
 	protected void forceComplete(MessageGroup group) {
 
 		Object correlationKey = group.getGroupId();
@@ -533,9 +544,8 @@ protected void forceComplete(MessageGroup group) {
 							expireGroup(correlationKey, groupNow);
 						}
 						if (!this.expireGroupsUponTimeout) {
-							afterRelease(groupNow, groupNow.getMessages());
+							afterRelease(groupNow, groupNow.getMessages(), true);
 							removeGroup = false;
-							this.messageStore.completeGroup(correlationKey); // late messages immediately discarded
 						}
 					}
 					else {

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

@@ -63,6 +63,11 @@ public void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) {
 
 	@Override
 	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
+		afterRelease(messageGroup, completedMessages, false);
+	}
+
+	@Override
+	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages, boolean timeout) {
 
 		int size = messageGroup.getMessages().size();
 		int sequenceSize = 0;
@@ -82,6 +87,9 @@ protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> co
 					this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), msg);
 				}
 			}
+			if (timeout) {
+				this.messageStore.completeGroup(messageGroup.getGroupId());
+			}
 		}
 	}