GH-3334: Add "embedded reaper" into CorrelationMH

Fixes #3334

* Add `expireTimeout` property into `AbstractCorrelatingMessageHandler`
to call newly introduced `purgeOrphanedGroups()` API for removing old
groups from the store
* Add `expireDuration` to perform `purgeOrphanedGroups()` task periodically

Author: artembilan

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.aggregator;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -53,6 +54,7 @@
 import org.springframework.integration.support.locks.DefaultLockRegistry;
 import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.integration.util.UUIDConverter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
@@ -82,6 +84,11 @@
  * Use proper {@link CorrelationStrategy} for cases when same
  * {@link org.springframework.integration.store.MessageStore} is used
  * for multiple handlers to ensure uniqueness of message groups across handlers.
+ * <p>
+ * When the {@link #expireTimeout} is more than 0, the groups which older then this timeout
+ * is purged from the store on start up (or when {@link #purgeOrphanedGroups()} is called).
+ * If {@link #expireDuration} is provide, the task is scheduled to perform
+ * {@link #purgeOrphanedGroups()} periodically.
  *
  * @author Iwein Fuld
  * @author Dave Syer
@@ -132,6 +139,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private List<Advice> forceReleaseAdviceChain;
 
+	private long expireTimeout;
+
+	@Nullable
+	private Duration expireDuration;
+
 	private MessageGroupProcessor forceReleaseProcessor = new ForceReleaseMessageGroupProcessor();
 
 	private EvaluationContext evaluationContext;
@@ -310,6 +322,32 @@ public void setReleaseLockBeforeSend(boolean releaseLockBeforeSend) {
 		this.releaseLockBeforeSend = releaseLockBeforeSend;
 	}
 
+	/**
+	 * Configure a timeout in milliseconds for purging old orphaned groups from the store.
+	 * Used on startup and if {@link #expireDuration} is provided the task for {@link #purgeOrphanedGroups()}
+	 * is scheduled with that period.
+	 * The {@link #forceReleaseProcessor} is used to process those expired groups according
+	 * the "force complete" options.
+	 * @param expireTimeout the number of milliseconds to determine old orphaned groups in the store to purge.
+	 * @since 5.4
+	 * @see #purgeOrphanedGroups()
+	 */
+	public void setExpireTimeout(long expireTimeout) {
+		Assert.isTrue(expireTimeout > 0, "'expireTimeout' must be more than 0.");
+		this.expireTimeout = expireTimeout;
+	}
+
+	/**
+	 * Configure a {@link Duration} how often to clean up old orphaned groups from the store.
+	 * @param expireDuration the delay how often to call {@link #purgeOrphanedGroups()}.
+	 * @since 5.4
+	 * @see #purgeOrphanedGroups()
+	 * @see #setExpireTimeout(long)
+	 */
+	public void setExpireDuration(@Nullable Duration expireDuration) {
+		this.expireDuration = expireDuration;
+	}
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.applicationEventPublisher = applicationEventPublisher;
@@ -896,6 +934,13 @@ public void start() {
 			if (this.releaseStrategy instanceof Lifecycle) {
 				((Lifecycle) this.releaseStrategy).start();
 			}
+			if (this.expireTimeout > 0) {
+				purgeOrphanedGroups();
+				if (this.expireDuration != null) {
+					getTaskScheduler()
+							.scheduleWithFixedDelay(this::purgeOrphanedGroups, this.expireDuration);
+				}
+			}
 		}
 	}
 
@@ -917,6 +962,17 @@ public boolean isRunning() {
 		return this.running;
 	}
 
+	/**
+	 * Perform a {@link MessageGroupStore#expireMessageGroups(long)} with the provided {@link #expireTimeout}.
+	 * Can be called externally at any time.
+	 * Internally it is called from the scheduled task with the configured {@link #expireDuration}.
+	 * @since 5.4
+	 */
+	public void purgeOrphanedGroups() {
+		Assert.isTrue(expireTimeout > 0, "'expireTimeout' must be more than 0.");
+		this.messageStore.expireMessageGroups(this.expireTimeout);
+	}
+
 	protected static class SequenceAwareMessageGroup extends SimpleMessageGroup {
 
 		private final SimpleMessageGroup sourceGroup;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -17,11 +17,14 @@
 package org.springframework.integration.aggregator;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
 import java.lang.reflect.Method;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -31,10 +34,11 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Ignore;
-import org.junit.Test;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.store.MessageGroup;
@@ -57,7 +61,7 @@
  */
 public class AbstractCorrelatingMessageHandlerTests {
 
-	@Test // INT-2751
+	@Test
 	public void testReaperDoesntReapAProcessingGroup() throws Exception {
 		final MessageGroupStore groupStore = new SimpleMessageStore();
 		final CountDownLatch waitForSendLatch = new CountDownLatch(1);
@@ -147,7 +151,7 @@ protected void afterRelease(MessageGroup group, Collection<Message<?>> completed
 		exec.shutdownNow();
 	}
 
-	@Test // INT-2833
+	@Test
 	public void testReaperReapsAnEmptyGroup() {
 		final MessageGroupStore groupStore = new SimpleMessageStore();
 		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
@@ -176,7 +180,7 @@ public void testReaperReapsAnEmptyGroup() {
 				.isEqualTo(0);
 	}
 
-	@Test // INT-2833
+	@Test
 	public void testReaperReapsAnEmptyGroupAfterConfiguredDelay() throws Exception {
 		final MessageGroupStore groupStore = new SimpleMessageStore();
 		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
@@ -251,7 +255,7 @@ public void testReapWithChangeInSameMillisecond() throws Exception {
 		assertThat(payload.size()).isEqualTo(1);
 	}
 
-	@Test /* INT-3216 */
+	@Test
 	public void testDontReapIfAlreadyComplete() throws Exception {
 		MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
 		AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
@@ -384,7 +388,7 @@ public void removeMessageGroup(Object groupId) {
 	}
 
 	@Test
-	@Ignore("Time sensitive: the empty group might be removed before main thread reaches assertion for size")
+	@Disabled("Time sensitive: the empty group might be removed before main thread reaches assertion for size")
 	public void testScheduleRemoveAnEmptyGroupAfterConfiguredDelay() throws Exception {
 		final MessageGroupStore groupStore = new SimpleMessageStore();
 		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
@@ -430,7 +434,7 @@ public void testScheduleRemoveAnEmptyGroupAfterConfiguredDelay() throws Exceptio
 	}
 
 	@Test
-	@Ignore("Until 5.2 with new 'owner' feature on groups")
+	@Disabled("Until 5.2 with new 'owner' feature on groups")
 	public void testDontReapMessageOfOtherHandler() {
 		MessageGroupStore groupStore = new SimpleMessageStore();
 
@@ -484,4 +488,45 @@ public void testNoPopSequenceDetails() {
 		assertThat(receive.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS)).isTrue();
 	}
 
+	@Test
+	public void testPurgeOrphanedGroupsOnStartup() throws InterruptedException {
+		MessageGroupStore groupStore = new SimpleMessageStore();
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
+		handler.setReleaseStrategy(group -> false);
+		QueueChannel discardChannel = new QueueChannel();
+		handler.setDiscardChannel(discardChannel);
+		handler.setExpireTimeout(1);
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.afterPropertiesSet();
+		handler.handleMessageInternal(MessageBuilder.withPayload("test").setCorrelationId("test").build());
+		Thread.sleep(100);
+		handler.start();
+		Message<?> receive = discardChannel.receive(10000);
+		assertThat(receive).isNotNull();
+		assertThat(groupStore.getMessageGroupCount()).isEqualTo(0);
+	}
+
+	@Test
+	public void testPurgeOrphanedGroupsScheduled() {
+		MessageGroupStore groupStore = spy(new SimpleMessageStore());
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
+		handler.setReleaseStrategy(group -> false);
+		QueueChannel discardChannel = new QueueChannel();
+		handler.setDiscardChannel(discardChannel);
+		handler.setExpireTimeout(100);
+		handler.setExpireDuration(Duration.ofMillis(10));
+		handler.setBeanFactory(mock(BeanFactory.class));
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		handler.setTaskScheduler(taskScheduler);
+		handler.afterPropertiesSet();
+		handler.handleMessageInternal(MessageBuilder.withPayload("test").setCorrelationId("test").build());
+		handler.start();
+		Message<?> receive = discardChannel.receive(10000);
+		assertThat(receive).isNotNull();
+		assertThat(groupStore.getMessageGroupCount()).isEqualTo(0);
+		verify(groupStore, atLeast(2)).expireMessageGroups(100);
+		taskScheduler.destroy();
+	}
+
 }