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 70461f4007..3941e99b8e 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 @@ -91,7 +91,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private final Comparator> sequenceNumberComparator = new SequenceNumberComparator(); - private final Map> expireGroupScheduledFutures = new HashMap>(); + private final Map> 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> completedMessages, boolean timeout) { + protected void afterRelease(MessageGroup group, Collection> completedMessages, boolean timeout) { afterRelease(group, completedMessages); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java index 0a53cd5c5b..de82cd3b93 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java +++ b/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> completedMessages) { Object groupId = messageGroup.getGroupId(); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd index 544156afd2..3241cbe0b3 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd @@ -3834,15 +3834,16 @@ - 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. 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 82ad134b74..8acc7b4d61 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 @@ -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 message = MessageBuilder.withPayload("foo") + Message 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 message = MessageBuilder.withPayload("foo") + Message 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 message = MessageBuilder.withPayload("foo") + Message 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> outputMessages = new ArrayList>(); + 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 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()); + } + } diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 6ff8e7dae8..b152740373 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/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 `` definitions. diff --git a/src/reference/asciidoc/resequencer.adoc b/src/reference/asciidoc/resequencer.adoc index 32ec905123..53d9492c67 100644 --- a/src/reference/asciidoc/resequencer.adoc +++ b/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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index f0aa4b4691..9e778bac0c 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -81,4 +81,5 @@ See <> 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 <> for more information.