INT-4091: Aggregator: Removal for Empty Groups

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

When `empty-group-min-timeout` is configured and `expireGroupsUponCompletion == false` and normal or partial sequences group release happens,
schedule the group for removal after `empty-group-min-timeout`, since it is empty already.
That lets to avoid `MessageGroupStoreReaper` configuration just for cleaning empty groups.

* Retrieve the group `lastModified` to check if group is still valid for removal
* Remove `ScheduledFeature` in the remove task
* Polishing for debug messages to reflect the current logic
* Reschedule empty group removal task in case of `InterruptedException` on the `lock.lockInterruptibly()`
* Fix typos in docs

Fix race condition between `groupStore.expireMessageGroups()`and `TaskScheduler`  in the `AbstractCorrelatingMessageHandlerTests.testReaperReapsAnEmptyGroupAfterConfiguredDelay()`

Also check groupSize for removal decision

Address PR comments

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

@@ -91,7 +91,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private final Comparator<Message<?>> sequenceNumberComparator = new SequenceNumberComparator();
 
-	private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<UUID, ScheduledFuture<?>>();
+	private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<>();
 
 	private final MessageGroupProcessor outputProcessor;
 
@@ -163,8 +163,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 	public final void setMessageStore(MessageGroupStore store) {
 		this.messageStore = store;
 		store.registerMessageGroupExpiryCallback(
-				(messageGroupStore, group) ->
-					AbstractCorrelatingMessageHandler.this.forceReleaseProcessor.processMessageGroup(group));
+				(messageGroupStore, group) -> this.forceReleaseProcessor.processMessageGroup(group));
 	}
 
 	public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
@@ -254,9 +253,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 		if (this.groupTimeoutExpression != null && !CollectionUtils.isEmpty(this.forceReleaseAdviceChain)) {
 			ProxyFactory proxyFactory = new ProxyFactory(processor);
-			for (Advice advice : this.forceReleaseAdviceChain) {
-				proxyFactory.addAdvice(advice);
-			}
+			this.forceReleaseAdviceChain.forEach(proxyFactory::addAdvice);
 			return (MessageGroupProcessor) proxyFactory.getProxy(getApplicationContext().getClassLoader());
 		}
 		return processor;
@@ -405,7 +402,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 			if (scheduledFuture != null) {
 				boolean canceled = scheduledFuture.cancel(true);
 				if (canceled && this.logger.isDebugEnabled()) {
-					this.logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ "
+					this.logger.debug("Cancel 'ScheduledFuture' for MessageGroup with Correlation Key [ "
 							+ correlationKey + "].");
 				}
 			}
@@ -430,6 +427,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 						// processing messages
 						this.afterRelease(messageGroup, completedMessages);
 					}
+					if (!isExpireGroupsUponCompletion() && this.minimumTimeoutForEmptyGroups > 0) {
+						removeEmptyGroupAfterTimeout(messageGroup, this.minimumTimeoutForEmptyGroups);
+					}
 				}
 				else {
 					scheduleGroupToForceComplete(messageGroup);
@@ -444,6 +444,57 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		}
 	}
 
+	protected boolean isExpireGroupsUponCompletion() {
+		return false;
+	}
+
+	private void removeEmptyGroupAfterTimeout(MessageGroup messageGroup, long timeout) {
+		Object groupId = messageGroup.getGroupId();
+		UUID groupUuid = UUIDConverter.getUUID(groupId);
+		ScheduledFuture<?> scheduledFuture = getTaskScheduler()
+				.schedule(() -> {
+					Lock lock = this.lockRegistry.obtain(groupUuid.toString());
+
+					try {
+						lock.lockInterruptibly();
+						try {
+							this.expireGroupScheduledFutures.remove(groupUuid);
+							/*
+							 * Obtain a fresh state for group from the MessageStore,
+							 * since it could be changed while we have waited for lock.
+							 */
+							MessageGroup groupNow = this.messageStore.getMessageGroup(groupUuid);
+							boolean removeGroup = groupNow.size() == 0 &&
+									groupNow.getLastModified()
+											<= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
+							if (removeGroup) {
+								if (this.logger.isDebugEnabled()) {
+									this.logger.debug("Removing empty group: " + groupUuid);
+								}
+								this.messageStore.removeMessageGroup(groupId);
+							}
+						}
+						finally {
+							lock.unlock();
+						}
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+						if (this.logger.isDebugEnabled()) {
+							this.logger.debug("Thread was interrupted while trying to obtain lock."
+									+ "Rescheduling empty MessageGroup [ " + groupId + "] for removal.");
+						}
+						removeEmptyGroupAfterTimeout(messageGroup, timeout);
+					}
+
+				}, new Date(System.currentTimeMillis() + timeout));
+
+		if (this.logger.isDebugEnabled()) {
+			this.logger.debug("Schedule empty MessageGroup [ " + groupId + "] for removal.");
+		}
+		this.expireGroupScheduledFutures.put(groupUuid, scheduledFuture);
+	}
+
 	private void scheduleGroupToForceComplete(MessageGroup messageGroup) {
 		final Long groupTimeout = obtainGroupTimeout(messageGroup);
 		/*
@@ -506,7 +557,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 	 * @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) {
+	protected void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages, boolean timeout) {
 		afterRelease(group, completedMessages);
 	}
 

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

@@ -61,6 +61,11 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler
 		this.expireGroupsUponCompletion = expireGroupsUponCompletion;
 	}
 
+	@Override
+	protected boolean isExpireGroupsUponCompletion() {
+		return this.expireGroupsUponCompletion;
+	}
+
 	@Override
 	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
 		Object groupId = messageGroup.getGroupId();

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd

@@ -3834,15 +3834,16 @@
 				<xsd:attribute name="empty-group-min-timeout" type="xsd:string">
 					<xsd:annotation>
 						<xsd:documentation>
-							Only applies if a MessageGroupStoreReaper is configured for this Correlation
-							Endpoint's MessageStore.
-							By default, when a MessageGroupStoreReaper is configured to expire partial
-							groups, empty groups are also removed. Empty groups exist after a group
-							is released normally. This is to enable the detection and discarding of
-							late-arriving messages. If you wish to run empty group deletion on a longer
+							Empty groups exist after a group is released normally.
+							This is to enable the detection and discarding of late-arriving messages.
+							If you wish to run empty group deletion on a longer
 							schedule than expiring partial groups, set this property. Empty groups will
 							then not be removed from the MessageStore until they have not been modified
 							for at least this number of milliseconds.
+							If this is set, the group is scheduled for removal after normal
+							or partial sequences group release.
+							When a MessageGroupStoreReaper is configured to expire partial
+							groups, empty groups are also removed, but using this value.
 							Note that the actual time to expire an
 							empty group will also be affected by the reaper's 'timeout'
 							property and it could be as much as this value plus the timeout.

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

@@ -47,6 +47,7 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 /**
  * @author Gary Russell
@@ -118,7 +119,7 @@ public class AbstractCorrelatingMessageHandlerTests {
 		handler.setDiscardChannel(discards);
 		handler.setSendPartialResultOnExpiry(true);
 
-		Message<String>	message = MessageBuilder.withPayload("foo")
+		Message<String> message = MessageBuilder.withPayload("foo")
 				.setCorrelationId("qux")
 				.build();
 		// partial group that will be reaped
@@ -159,7 +160,7 @@ public class AbstractCorrelatingMessageHandlerTests {
 		});
 		handler.setReleaseStrategy(group -> group.size() == 1);
 
-		Message<String>	message = MessageBuilder.withPayload("foo")
+		Message<String> message = MessageBuilder.withPayload("foo")
 				.setCorrelationId("bar")
 				.build();
 		handler.handleMessage(message);
@@ -186,20 +187,32 @@ public class AbstractCorrelatingMessageHandlerTests {
 		});
 		handler.setReleaseStrategy(group -> group.size() == 1);
 
-		handler.setMinimumTimeoutForEmptyGroups(1000);
-
-		Message<String>	message = MessageBuilder.withPayload("foo")
+		Message<String> message = MessageBuilder.withPayload("foo")
 				.setCorrelationId("bar")
 				.build();
 		handler.handleMessage(message);
 
+		handler.setMinimumTimeoutForEmptyGroups(100);
+
 		assertEquals(1, outputMessages.size());
 
 		assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
 		groupStore.expireMessageGroups(0);
 		assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
-		Thread.sleep(1010);
-		groupStore.expireMessageGroups(0);
+
+		int n = 0;
+
+		while (n++ < 200) {
+			groupStore.expireMessageGroups(0);
+			if (TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size() > 0) {
+				Thread.sleep(50);
+			}
+			else {
+				break;
+			}
+		}
+
+		assertTrue(n < 200);
 		assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
 	}
 
@@ -321,6 +334,7 @@ public class AbstractCorrelatingMessageHandlerTests {
 		handler.setReleaseStrategy(group -> true);
 		handler.setExpireGroupsUponTimeout(false);
 		SimpleMessageStore messageStore = new SimpleMessageStore() {
+
 			@Override
 			public void removeMessageGroup(Object groupId) {
 				throw new RuntimeException("intentional");
@@ -354,8 +368,51 @@ public class AbstractCorrelatingMessageHandlerTests {
 		/* Since MessageGroup had been marked as 'complete', but hasn't been removed because of exception,
 		 the second message is discarded
 		  */
-		Message<?> receive = discardChannel.receive(1000);
+		Message<?> receive = discardChannel.receive(10000);
 		assertNotNull(receive);
 	}
 
+	@Test
+	public void testScheduleRemoveAnEmptyGroupAfterConfiguredDelay() throws Exception {
+		final MessageGroupStore groupStore = new SimpleMessageStore();
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
+
+		final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
+		handler.setOutputChannel((message, timeout) -> {
+			/*
+			 * Executes when group 'bar' completes normally
+			 */
+			outputMessages.add(message);
+			return true;
+		});
+		handler.setReleaseStrategy(group -> group.size() == 1);
+
+		handler.setMinimumTimeoutForEmptyGroups(100);
+
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		handler.setTaskScheduler(taskScheduler);
+
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setCorrelationId("bar")
+				.build();
+		handler.handleMessage(message);
+
+		assertEquals(1, outputMessages.size());
+
+		assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
+
+		Thread.sleep(100);
+
+		int n = 0;
+
+		while (TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size() > 0
+				&& n++ < 200) {
+			Thread.sleep(50);
+		}
+
+		assertTrue(n < 200);
+		assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
+	}
+
 }

src/reference/asciidoc/aggregator.adoc

@@ -505,6 +505,9 @@ Since _version 4.1_, you can now control this behavior using `expire-groups-upon
 NOTE: When a group is timed out, the `ReleaseStrategy` is given one more opportunity to release the group; if it does so, and `expire-groups-upon-timeout` is false, then expiration is controlled by `expire-groups-upon-completion`.
 If the group is not released by the release strategy during timeout, then the expiration is controlled by the `expire-groups-upon-timeout`.
 Timed-out groups are either discarded, or a partial release occurs (based on `send-partial-result-on-expiry`).
+
+Starting with _version 5.0_ empty groups are also scheduled for removal after `empty-group-min-timeout`.
+If `expireGroupsUponCompletion == false` and `minimumTimeoutForEmptyGroups > 0`, the task to remove the group is scheduled, when normal or partial sequences release happens.
 =====
 
 Using a `ref` attribute is generally recommended if a custom aggregator handler implementation may be referenced in other `<aggregator>` definitions.

src/reference/asciidoc/resequencer.adoc

@@ -161,6 +161,7 @@ Late arriving messages will be immediately discarded.
 Set this to `true` to remove the group completely; then, late arriving messages will start a new group and won't be discarded until the group again times out.
 The new group will never be released normally because of the "hole" in the sequence range that caused the timeout.
 Empty groups can be expired (completely removed) later using a `MessageGroupStoreReaper` together with the `empty-group-min-timeout` attribute.
+Starting with _version 5.0_ empty groups are also scheduled for removal after `empty-group-min-timeout`.
 Default: 'false'.
 
 NOTE: Since there is no custom behavior to be implemented in Java classes for resequencers, there is no annotation support for it.

src/reference/asciidoc/whats-new.adoc

@@ -81,4 +81,5 @@ See <<http-header-mapping>> for more information.
 ==== Aggregator Performance Changes
 
 Aggregators now use a `SimpleSequenceSizeReleaseStrategy` by default, which is more efficient, especially with large groups.
+Empty groups are now scheduled for removal after `empty-group-min-timeout`.
 See <<aggregator>> for more information.