diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 7bac314de8..eea9ceaf76 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/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. + *

+ * 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 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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index b26789b521..c0e355f978 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/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> 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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index da824b050b..7d03453310 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/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"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index f98603b5b7..7a9295c206 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/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 + + + + 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. + + + + + + + The 'Duration' (in milliseconds) how often to run purge old groups scheduled task. + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index 2df62053e7..9442507aad 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/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(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml index 9d472ccb30..6ed6b5c754 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml +++ b/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"> diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index c36e625929..39ddc1e189 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/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 diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 286a29cbd6..5aadf1cbd4 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -513,6 +513,7 @@ Only this sub-element or `` 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,18 @@ 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 with version 5.4, the aggregator (and resequencer) can be configured to expire orphaned groups (groups in a persistent message store that might not otherwise be released). +The `expireTimeout` (if greater than `0`) indicates that groups older than this value 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 to the `forceComplete(MessageGroup)` functionality according to 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 restart, when using a persistent message group store. +The functionality is similar to the `MessageGroupStoreReaper` with a scheduled task, but provides a convenient way to deal with old groups within specific components, when using group timeout instead of a reaper. +The `MessageGroupStore` must be provided exclusively for the current correlation endpoint. +Otherwise one aggregator may purge groups from another. +With the aggregator, groups expired using this technique will either be discarded or released as a partial group, depending on the `expireGroupsUponCompletion` property. ===== We generally recommend using a `ref` attribute if a custom aggregator handler implementation may be referenced in other `` definitions. diff --git a/src/reference/asciidoc/resequencer.adoc b/src/reference/asciidoc/resequencer.adoc index 9fb19dab3a..2732103360 100644 --- a/src/reference/asciidoc/resequencer.adoc +++ b/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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 0ba56c62f3..5f8ab71f28 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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) can now expire orphaned groups (groups in a persistent store where no new messages arrive after an application restart). +See <<./aggregator.adoc#aggregator-expiring-groups, Aggregator Expiring Groups>> for more information.