GH-3334: Add "embedded reaper" into CorrelationMH (#3342)

* GH-3334: Add "embedded reaper" into CorrelationMH

Fixes https://github.com/spring-projects/spring-integration/issues/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

* * Add Java DSL and XML support for `expireTimeout` and `expireDuration` options
* Document the new feature

* * Fix language in docs

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

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.AbstractIntegrationMessageBuilder
 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 @@ import org.springframework.util.ObjectUtils;
  * 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 greater than 0, groups which are older than this timeout
+ * are purged from the store on start up (or when {@link #purgeOrphanedGroups()} is called).
+ * If {@link #expireDuration} is provided, 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,45 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		this.releaseLockBeforeSend = releaseLockBeforeSend;
 	}
 
+	/**
+	 * Configure a timeout in milliseconds for purging old orphaned groups from the store.
+	 * Used on startup and when an {@link #expireDuration} is provided, the task for running
+	 * {@link #purgeOrphanedGroups()} is scheduled with that period.
+	 * The {@link #forceReleaseProcessor} is used to process those expired groups according
+	 * the "force complete" options. A group can be orphaned if a persistent message group
+	 * store is used and no new messages arrive for that group after a restart.
+	 * @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} (in millis) 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 #setExpireDuration(Duration)
+	 * @see #setExpireTimeout(long)
+	 */
+	public void setExpireDurationMillis(long expireDuration) {
+		setExpireDuration(Duration.ofMillis(expireDuration));
+	}
+
+	/**
+	 * 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 +947,13 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 			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 +975,17 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		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(this.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/main/java/org/springframework/integration/config/AggregatorFactoryBean.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.config;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -62,13 +63,6 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 
 	private String outputChannelName;
 
-	@SuppressWarnings("deprecation")
-	private org.springframework.integration.support.management.AbstractMessageHandlerMetrics metrics;
-
-	private Boolean statsEnabled;
-
-	private Boolean countsEnabled;
-
 	private LockRegistry lockRegistry;
 
 	private MessageGroupStore messageStore;
@@ -97,6 +91,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 
 	private Boolean releaseLockBeforeSend;
 
+	private Long expireTimeout;
+
+	private Long expireDuration;
+
 	private Function<MessageGroup, Map<String, Object>> headersFunction;
 
 	public void setProcessorBean(Object processorBean) {
@@ -120,25 +118,6 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 		this.outputChannelName = outputChannelName;
 	}
 
-	/**
-	 * Deprecated.
-	 * @param metrics the metrics.
-	 * @deprecated in favor of Micrometer metrics.
-	 */
-	@Deprecated
-	@SuppressWarnings("deprecation")
-	public void setMetrics(org.springframework.integration.support.management.AbstractMessageHandlerMetrics metrics) {
-		this.metrics = metrics;
-	}
-
-	public void setStatsEnabled(Boolean statsEnabled) {
-		this.statsEnabled = statsEnabled;
-	}
-
-	public void setCountsEnabled(Boolean countsEnabled) {
-		this.countsEnabled = countsEnabled;
-	}
-
 	public void setLockRegistry(LockRegistry lockRegistry) {
 		this.lockRegistry = lockRegistry;
 	}
@@ -199,7 +178,14 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 		this.headersFunction = headersFunction;
 	}
 
-	@SuppressWarnings("deprecation")
+	public void setExpireTimeout(Long expireTimeout) {
+		this.expireTimeout = expireTimeout;
+	}
+
+	public void setExpireDurationMillis(Long expireDuration) {
+		this.expireDuration = expireDuration;
+	}
+
 	@Override
 	protected AggregatingMessageHandler createHandler() {
 		MessageGroupProcessor outputProcessor;
@@ -229,9 +215,6 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 				.acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion)
 				.acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout)
 				.acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName)
-				.acceptIfNotNull(this.metrics, aggregator::configureMetrics)
-				.acceptIfNotNull(this.statsEnabled, aggregator::setStatsEnabled)
-				.acceptIfNotNull(this.countsEnabled, aggregator::setCountsEnabled)
 				.acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry)
 				.acceptIfNotNull(this.messageStore, aggregator::setMessageStore)
 				.acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy)
@@ -245,7 +228,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 				.acceptIfNotNull(this.minimumTimeoutForEmptyGroups, aggregator::setMinimumTimeoutForEmptyGroups)
 				.acceptIfNotNull(this.expireGroupsUponTimeout, aggregator::setExpireGroupsUponTimeout)
 				.acceptIfNotNull(this.popSequence, aggregator::setPopSequence)
-				.acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend);
+				.acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend)
+				.acceptIfNotNull(this.expireDuration,
+						(duration) -> aggregator.setExpireDuration(Duration.ofMillis(duration)))
+				.acceptIfNotNull(this.expireTimeout, aggregator::setExpireTimeout);
 
 		return aggregator;
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.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.
@@ -101,6 +101,9 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
 
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_TIMEOUT);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_LOCK);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expire-timeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expire-duration",
+				"expireDurationMillis");
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
@@ -329,4 +330,28 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
 		return _this();
 	}
 
+	/**
+	 * Configure a timeout for old groups in the store to purge.
+	 * @param expireTimeout the timeout in milliseconds to use.
+	 * @return the endpoint spec.
+	 * @since 5.4
+	 * @see AbstractCorrelatingMessageHandler#setExpireTimeout(long)
+	 */
+	public S setExpireTimeout(long expireTimeout) {
+		this.handler.setExpireTimeout(expireTimeout);
+		return _this();
+	}
+
+	/**
+	 * Configure a {@link Duration} how often to run a scheduled purge task.
+	 * @param expireDuration the duration for scheduled purge task.
+	 * @return the endpoint spec.
+	 * @since 5.4
+	 * @see AbstractCorrelatingMessageHandler#setExpireDuration(Duration)
+	 */
+	public S setExpireDuration(Duration expireDuration) {
+		this.handler.setExpireDuration(expireDuration);
+		return _this();
+	}
+
 }

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

@@ -4037,6 +4037,22 @@
 						<xsd:union memberTypes="xsd:boolean xsd:string"/>
 					</xsd:simpleType>
 				</xsd:attribute>
+				<xsd:attribute name="expire-timeout" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							The timeout for old groups in the store to purge on startup.
+							If 'expire-duration' is also provided, the purge task is scheduled
+							periodically to purge old groups with this expiration timeout.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="expire-duration" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							The 'Duration' (in milliseconds) how often to run purge old groups scheduled task.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

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.ExecutorService;
 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 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  */
 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 @@ public class AbstractCorrelatingMessageHandlerTests {
 		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 class AbstractCorrelatingMessageHandlerTests {
 				.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 class AbstractCorrelatingMessageHandlerTests {
 		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 class AbstractCorrelatingMessageHandlerTests {
 	}
 
 	@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 class AbstractCorrelatingMessageHandlerTests {
 	}
 
 	@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 class AbstractCorrelatingMessageHandlerTests {
 		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();
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml

@@ -56,7 +56,9 @@
 		scheduler="scheduler"
 		message-store="store"
 		pop-sequence="false"
-		order="5">
+		order="5"
+		expire-duration="10000"
+		expire-timeout="250">
 			<expire-transactional/>
 	</aggregator>
 

spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.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.
@@ -19,13 +19,13 @@ package org.springframework.integration.config;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanCreationException;
@@ -54,7 +54,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Marius Bogoevici
@@ -65,7 +65,7 @@ import org.springframework.test.context.junit4.SpringRunner;
  * @author Gunnar Hillert
  * @author Gary Russell
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 public class AggregatorParserTests {
 
 	@Autowired
@@ -194,6 +194,9 @@ public class AggregatorParserTests {
 		assertThat(TestUtils.getPropertyValue(consumer, "order")).isEqualTo(5);
 		assertThat(TestUtils.getPropertyValue(consumer, "forceReleaseAdviceChain")).isNotNull();
 		assertThat(TestUtils.getPropertyValue(consumer, "popSequence", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(consumer, "expireTimeout")).isEqualTo(250L);
+		assertThat(TestUtils.getPropertyValue(consumer, "expireDuration", Duration.class))
+				.isEqualTo(Duration.ofSeconds(10));
 	}
 
 	@Test