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 b421879bd6..d589a2e920 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 @@ -92,6 +92,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH private boolean lockRegistrySet = false; + private volatile long minimumTimeoutForEmptyGroups; + public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor); @@ -172,6 +174,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH this.sendPartialResultOnExpiry = sendPartialResultOnExpiry; } + /** + * 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 + * 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. + * + * @param minimumTimeoutForEmptyGroups The minimum timeout. + */ + public void setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups) { + this.minimumTimeoutForEmptyGroups = minimumTimeoutForEmptyGroups; + } + public void setReleasePartialSequences(boolean releasePartialSequences){ Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy, "Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() @@ -241,7 +258,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH */ protected abstract void afterRelease(MessageGroup group, Collection> completedMessages); - private final boolean forceComplete(MessageGroup group) { + private void forceComplete(MessageGroup group) { Object correlationKey = group.getGroupId(); // UUIDConverter is no-op if already converted @@ -250,41 +267,47 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH try { lock.lockInterruptibly(); try { - if (group.size() > 0) { - try { - /* - * Need to verify the group hasn't changed while we were waiting on - * its lock. We have to re-fetch the group for this. A possible - * future improvement would be to add MessageGroupStore.getLastModified(groupId). - */ - MessageGroup messageGroupNow = this.messageStore.getMessageGroup( - group.getGroupId()); - long lastModifiedNow = messageGroupNow.getLastModified(); - if (group.getLastModified() == lastModifiedNow) { - if (releaseStrategy.canRelease(group)) { - this.completeGroup(correlationKey, group); - } - else { - this.expireGroup(correlationKey, group); - } + /* + * Need to verify the group hasn't changed while we were waiting on + * its lock. We have to re-fetch the group for this. A possible + * future improvement would be to add MessageGroupStore.getLastModified(groupId). + */ + MessageGroup messageGroupNow = this.messageStore.getMessageGroup( + group.getGroupId()); + long lastModifiedNow = messageGroupNow.getLastModified(); + if (group.getLastModified() == lastModifiedNow) { + if (group.size() > 0) { + if (releaseStrategy.canRelease(group)) { + this.completeGroup(correlationKey, group); } else { - removeGroup = false; - if (logger.isDebugEnabled()) { - logger.debug("Group expiry candidate (" + group.getGroupId() + - ") has changed - it may be reconsidered for a future expiration"); - } + this.expireGroup(correlationKey, group); } } - finally { - if (removeGroup) { - this.remove(group); + else { + /* + * By default empty groups are removed on the same schedule as non-empty + * groups. A longer timeout for empty groups can be enabled by + * setting minimumTimeoutForEmptyGroups. + */ + removeGroup = lastModifiedNow < (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups); + if (removeGroup && logger.isDebugEnabled()) { + logger.debug("Removing empty group: " + group.getGroupId()); } } - return true; + } + else { + removeGroup = false; + if (logger.isDebugEnabled()) { + logger.debug("Group expiry candidate (" + group.getGroupId() + + ") has changed - it may be reconsidered for a future expiration"); + } } } finally { + if (removeGroup) { + this.remove(group); + } lock.unlock(); } } @@ -292,7 +315,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH Thread.currentThread().interrupt(); throw new MessagingException("Thread was interrupted while trying to obtain lock"); } - return false; } void remove(MessageGroup group) { 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 5a3d1d46dd..6ef72047d1 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 @@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -34,6 +35,7 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; /** * @author Gary Russell @@ -150,4 +152,101 @@ public class AbstractCorrelatingMessageHandlerTests { assertNull(discards.receive(0)); } + @Test // INT-2833 + public void testReaperReapsAnEmptyGroup() throws Exception { + final MessageGroupStore groupStore = new SimpleMessageStore(); + AggregatingMessageHandler handler = new AggregatingMessageHandler( + new MessageGroupProcessor() { + + public Object processMessageGroup(MessageGroup group) { + return group; + } + }, groupStore) { + }; + + final List> outputMessages = new ArrayList>(); + handler.setOutputChannel(new MessageChannel() { + + /* + * Executes when group 'bar' completes normally + */ + public boolean send(Message message, long timeout) { + outputMessages.add(message); + return true; + } + + public boolean send(Message message) { + return this.send(message, 0); + } + }); + handler.setReleaseStrategy(new ReleaseStrategy() { + + public boolean canRelease(MessageGroup group) { + return group.size() == 1; + } + }); + + 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()); + groupStore.expireMessageGroups(0); + assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size()); + } + + @Test // INT-2833 + public void testReaperReapsAnEmptyGroupAfterConfiguredDelay() throws Exception { + final MessageGroupStore groupStore = new SimpleMessageStore(); + AggregatingMessageHandler handler = new AggregatingMessageHandler( + new MessageGroupProcessor() { + + public Object processMessageGroup(MessageGroup group) { + return group; + } + }, groupStore) { + }; + + final List> outputMessages = new ArrayList>(); + handler.setOutputChannel(new MessageChannel() { + + /* + * Executes when group 'bar' completes normally + */ + public boolean send(Message message, long timeout) { + outputMessages.add(message); + return true; + } + + public boolean send(Message message) { + return this.send(message, 0); + } + }); + handler.setReleaseStrategy(new ReleaseStrategy() { + + public boolean canRelease(MessageGroup group) { + return group.size() == 1; + } + }); + + handler.setMinimumTimeoutForEmptyGroups(1000); + + 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()); + groupStore.expireMessageGroups(0); + assertEquals(1, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size()); + Thread.sleep(1010); + groupStore.expireMessageGroups(0); + assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size()); + } + } diff --git a/src/reference/docbook/aggregator.xml b/src/reference/docbook/aggregator.xml index 63653368d8..529db8b0d3 100644 --- a/src/reference/docbook/aggregator.xml +++ b/src/reference/docbook/aggregator.xml @@ -31,8 +31,9 @@ release. Correlation determines how messages are grouped for aggregation. - In Spring Integration correlation is done by default based on the CORRELATION_ID message - header. Messages with the same CORRELATION_ID will be grouped + In Spring Integration correlation is done by default based on the + MessageHeaders.CORRELATION_ID message + header. Messages with the same MessageHeaders.CORRELATION_ID will be grouped together. However, the correlation strategy may be customized to allow other ways of specifying how the messages should be grouped together by implementing a CorrelationStrategy (see below). @@ -40,7 +41,8 @@ To determine the point at which a group of messages is ready to be processed, a ReleaseStrategy is consulted. The default release strategy for the Aggregator will release a group when all - messages included in a sequence are present, based on the SEQUENCE_SIZE header. + messages included in a sequence are present, based on the + MessageHeaders.SEQUENCE_SIZE header. This default strategy may be overridden by providing a reference to a custom ReleaseStrategy implementation. @@ -70,9 +72,10 @@
- CorrelatingMessageHandler + AggregatingMessageHandler - The CorrelatingMessageHandler is a + The AggregatingMessageHandler (subclass of + AbstractCorrelatingMessageHandler) is a MessageHandler implementation, encapsulating the common functionalities of an Aggregator (and other correlating use cases), which are: @@ -120,9 +123,9 @@ The CorrelationStrategy is owned by the - CorrelatingMessageHandler + AbstractCorrelatingMessageHandler - and it has a default value based on the CORRELATION_ID message header: + and it has a default value based on the MessageHeaders.CORRELATION_ID message header: -As you can see based on the above signatures, the POJO-based Release Strategy will be passed a Collection of unmarked Messages -if you need access to the whole Message or Collection of payload objects if the type parameter is -anything other than Message. Typically this would satisfy the majority of use cases. However if -for some reason you need to access the full MessageGroup - which contains unmarked and marked Messages - +As you can see based on the above signatures, the POJO-based Release Strategy will be passed a +Collection of not-yet-released Messages +(if you need access to the whole Message) or a Collection of payload objects +(if the type parameter is +anything other than Message). Typically this would satisfy the majority of use cases. However if, +for some reason, you need to access the full MessageGroup then you should simply provide an implementation of the ReleaseStrategy interface. - When the group is released for aggregation, all its unmarked - messages are processed and then marked so they will not be processed - again. If the group is also complete (i.e. if all messages from a + When the group is released for aggregation, all its not-yet-released + messages are processed and removed from the group. + If the group is also complete (i.e. if all messages from a sequence have arrived or if there is no sequence defined), then the group - is removed from the message store. Partial sequences can be released, in - which case the next time the ReleaseStrategy is called it - will be presented with a group containing marked messages (already - processed) and unmarked messages (potentially a new partial - sequence). + is marked as complete. Any new messages for this group will be sent to the discard channel + (if defined). Setting expire-groups-upon-completion to true (default + is false) removes the entire group and any new messages, with the same correlation id + as the removed group, will form a new group. + Partial sequences can be released by using a MessageGroupStoreReaper + together with send-partial-result-on-expiry being set to true. + + To facilitate discarding of late-arriving messages, the aggregator must maintain state about + the group after it has been released. This can eventually cause out of memory conditions. + To avoid such situations, you should consider configuring a MessageGroupStoreReaper + to remove the group metadata; the expiry parameters should be set to expire groups after it is not + expected that late messages will arrive. For information about configuring a reaper, see + . Spring Integration provides an out-of-the box implementation for ReleaseStrategy, the @@ -338,7 +351,9 @@ then you should simply provide an implementation of the ReleaseStrate method="aggregate" ]]> ]]> ]]> @@ -458,6 +473,14 @@ then you should simply provide an implementation of the ReleaseStrate present). + + When set to true (default false), completed groups are + removed from the message store, allowing subsequent messages with + the same correlation to form a new group. The default behavior + is to send messages with the same correlation as a completed + group to the discard-channel. + + Using a ref attribute is generally recommended if a custom @@ -656,7 +679,7 @@ then you should simply provide an implementation of the ReleaseStrate
-
+
Managing State in an Aggregator: MessageGroupStore