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 56fdb89581..f4e4f15589 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 @@ -21,8 +21,10 @@ import java.util.Collections; import java.util.Comparator; import java.util.Date; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.Lock; @@ -75,6 +77,9 @@ import org.springframework.util.CollectionUtils; * By default the {@link CorrelationStrategy} will be a * {@link HeaderAttributeCorrelationStrategy} and the {@link ReleaseStrategy} will be a * {@link SequenceSizeReleaseStrategy}. + *

+ * Use proper {@link CorrelationStrategy} for cases when same {@link MessageStore} is used + * for multiple handlers to ensure uniqueness of message groups across handlers. * * @author Iwein Fuld * @author Dave Syer @@ -83,7 +88,7 @@ import org.springframework.util.CollectionUtils; * @author Artem Bilan * @author David Liu * @author Enrique Rodriguez - * + * @author Meherzad Lahewala * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler @@ -95,6 +100,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private final Map> expireGroupScheduledFutures = new HashMap<>(); + private final Set groupIds = new HashSet<>(); + private MessageGroupProcessor outputProcessor; private volatile MessageGroupStore messageStore; @@ -481,7 +488,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.logger.isDebugEnabled()) { this.logger.debug("Removing empty group: " + groupUuid); } - this.messageStore.removeMessageGroup(groupId); + remove(messageGroup); } } finally { @@ -671,9 +678,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } - void remove(MessageGroup group) { + protected void remove(MessageGroup group) { Object correlationKey = group.getGroupId(); this.messageStore.removeMessageGroup(correlationKey); + this.groupIds.remove(group.getGroupId()); } protected int findLastReleasedSequenceNumber(Object groupId, Collection> partialSequence) { @@ -682,6 +690,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } protected MessageGroup store(Object correlationKey, Message message) { + this.groupIds.add(correlationKey); return this.messageStore.addMessageToGroup(correlationKey, message); } @@ -851,7 +860,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override public Object processMessageGroup(MessageGroup group) { - forceComplete(group); + if (AbstractCorrelatingMessageHandler.this.groupIds.contains(group.getGroupId())) { + forceComplete(group); + } return null; } 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 e26745c41b..dd9f6d805e 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-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -52,6 +52,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell * @author Artem Bilan + * @author Meherzad Lahewala + * * @since 2.2 * */ @@ -417,4 +419,29 @@ public class AbstractCorrelatingMessageHandlerTests { assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size()); } + @Test + public void testDontReapMessageOfOtherHandler() throws Exception { + MessageGroupStore groupStore = new SimpleMessageStore(); + + AggregatingMessageHandler handler1 = new AggregatingMessageHandler(group -> group, groupStore); + AggregatingMessageHandler handler2 = new AggregatingMessageHandler(group -> group, groupStore); + + QueueChannel handler1DiscardChannel = new QueueChannel(); + handler1.setDiscardChannel(handler1DiscardChannel); + + QueueChannel handler2DiscardChannel = new QueueChannel(); + handler2.setDiscardChannel(handler2DiscardChannel); + + handler1.setReleaseStrategy(group -> false); + handler2.setReleaseStrategy(group -> false); + + handler1.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); + handler1.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); + handler2.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("bar").build()); + + groupStore.expireMessageGroups(0); + + assertTrue(handler1DiscardChannel.getQueueSize() == 2); + assertTrue(handler2DiscardChannel.getQueueSize() == 1); + } } diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index e574a90ac7..c3bb5e20e1 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -846,8 +846,8 @@ If the flag is set to true, then when the expiry callback is invoked, any unmark [IMPORTANT] ===== -When using a `MessageGroupStoreReaper`, it is generally recommended to use a separate `MessageStore` for each correlating endpoint. -Otherwise, unexpected results may occur because one endpoint may remove another endpoint's groups. +When a shared `MessageStore` is used for different correlation endpoints, it is necessary to configure a proper `CorrelationStrategy` to ensure uniqueness for group ids. +Otherwise unexpected behavior may happen when one correlation endpoint may release or expire messages from others - messages with the same correlation key are stored in the same message group. Some `MessageStore` implementations allow using the same physical resources, by partitioning the data; for example, the `JdbcMessageStore` has a `region` property; the `MongoDbMessageStore` has a `collectionName` property.