* Add Java DSL and XML support for expireTimeout and expireDuration options

* Document the new feature

Author: artembilan

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

@@ -85,9 +85,9 @@
  * {@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
+ * 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
@@ -324,10 +324,11 @@ public void setReleaseLockBeforeSend(boolean 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.
+	 * 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.
+	 * 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()
@@ -337,6 +338,18 @@ public void setExpireTimeout(long expireTimeout) {
 		this.expireTimeout = expireTimeout;
 	}
 
+	/**
+	 * Configure a {@link Duration} (ini 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()}.
@@ -969,7 +982,7 @@ public boolean isRunning() {
 	 * @since 5.4
 	 */
 	public void purgeOrphanedGroups() {
-		Assert.isTrue(expireTimeout > 0, "'expireTimeout' must be more than 0.");
+		Assert.isTrue(this.expireTimeout > 0, "'expireTimeout' must be more than 0.");
 		this.messageStore.expireMessageGroups(this.expireTimeout);
 	}
 

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 void setOutputChannelName(String outputChannelName) {
 		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 void setHeadersFunction(Function<MessageGroup, Map<String, Object>> heade
 		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 @@ protected AggregatingMessageHandler createHandler() {
 				.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 @@ protected AggregatingMessageHandler createHandler() {
 				.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 @@ protected void doParse(BeanDefinitionBuilder builder, Element element, BeanMetad
 
 		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 S popSequence(boolean popSequence) {
 		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/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 @@
 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,6 +54,7 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
@@ -65,7 +66,7 @@
  * @author Gunnar Hillert
  * @author Gary Russell
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 public class AggregatorParserTests {
 
 	@Autowired
@@ -194,6 +195,9 @@ public void testPropertyAssignment() {
 		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

src/reference/asciidoc/aggregator.adoc

@@ -513,6 +513,7 @@ Only this sub-element or `<expire-transactional/>` is allowed.
 A transaction `Advice` can also be configured here by using the Spring `tx` namespace.
 ====
 
+[[aggregator-expiring-groups]]
 [IMPORTANT]
 .Expiring Groups
 =====
@@ -535,6 +536,17 @@ Timed-out groups are either discarded or a partial release occurs (based on `sen
 
 Since 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.
+
+Starting version 5.4, the aggregator (and resequencer) can be configured for so-called "embedded reaper".
+The `expireTimeout` (if greater than `0`) indicates how old groups in the store should be purged.
+The `purgeOrphanedGroups()` method is called on start up and together with the provided `expireDuration` periodically within a scheduled task.
+This method is also can be called externally at any time.
+The expiration logic is fully delegated into a `forceComplete(MessageGroup)` functionality according the provided expiration options mentioned above.
+Such a periodic purge functionality is useful when a message store is needed to be cleaned up from those old groups which are not going to be released any more with regular message arrival logic.
+In most cases this happens after an application start up.
+The functionality is similar to the `MessageGroupStoreReaper` with a scheduled task, but provides a convenient way to deal with old groups withing specific components.
+The `MessageGroupStore` is still have to be provided exclusively for the current correlation endpoint.
+Otherwise one aggregator may purge groups from another.
 =====
 
 We generally recommend using a `ref` attribute if a custom aggregator handler implementation may be referenced in other `<aggregator>` definitions.

src/reference/asciidoc/resequencer.adoc

@@ -121,4 +121,6 @@ Starting with version 5.0, empty groups are also scheduled for removal after the
 The default is 'false'.
 ====
 
+Also see <<./aggregator.adoc#aggregator-expiring-groups, Aggregator Expiring Groups>> for more information.
+
 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

@@ -50,3 +50,6 @@ See <<./ip.adoc#ip-collaborating-adapters,Collaborating Channel Adapters>> and <
 
 The one-way messaging gateway (the `void` method return type) now sets a `nullChannel` explicitly into the `replyChannel` header to ignore any possible downstream replies.
 See <<./gateway.adoc#gateway-default-reply-channel,Setting the Default Reply Channel>> for more information.
+
+The aggregator (and resequencer) has now an "embedded reaper" support functionality provided via an `expireTimeout` and `expireDuration` options.
+See <<./aggregator.adoc#aggregator-expiring-groups, Aggregator Expiring Groups>> for more information.