From 5bf6161112373d720ef2b30a2f1a6776f4e42f55 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 15 Nov 2018 14:56:14 -0500 Subject: [PATCH] INT-4550: Disallow multi aggregators on same MGS (#2622) * INT-4550: Disallow multi aggregators on same MGS JIRA: https://jira.spring.io/browse/INT-4550 **Cherry-pick to 5.0.x** * * Introduce `UniqueExpiryCallback` * Use `UniqueExpiryCallback` in the `AbstractCorrelatingMessageHandler` * Check for uniqueness in the `AbstractMessageGroupStore` * Remove duplicate code in the `ConfigurableMongoDbMessageStore` * * Fix tests according a new logic * * Address PR review * Change `Assert.isTrue` to the `logger.error` for backward compatibility * Revert changes in tests since we don't throw exception anymore * Fix language on doc * * Fix Checkstyle violation in the `AbstractMessageGroupStore` * * Ignore `testDontReapMessageOfOtherHandler()` --- .../AbstractCorrelatingMessageHandler.java | 15 +-- .../store/AbstractMessageGroupStore.java | 18 ++- .../store/UniqueExpiryCallback.java | 33 +++++ ...bstractCorrelatingMessageHandlerTests.java | 1 + .../ConfigurableMongoDbMessageStore.java | 117 ++---------------- src/reference/asciidoc/aggregator.adoc | 7 ++ 6 files changed, 71 insertions(+), 120 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/store/UniqueExpiryCallback.java 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 e09868d05a..1daadb72d4 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 @@ -22,7 +22,6 @@ import java.util.Comparator; import java.util.Date; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; @@ -51,6 +50,7 @@ import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.integration.store.UniqueExpiryCallback; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.locks.DefaultLockRegistry; @@ -102,8 +102,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private final Map> expireGroupScheduledFutures = new ConcurrentHashMap<>(); - private final Set groupIds = ConcurrentHashMap.newKeySet(); - private MessageGroupProcessor outputProcessor; private MessageGroupStore messageStore; @@ -188,8 +186,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP public final void setMessageStore(MessageGroupStore store) { this.messageStore = store; - store.registerMessageGroupExpiryCallback( - (messageGroupStore, group) -> this.forceReleaseProcessor.processMessageGroup(group)); + UniqueExpiryCallback expiryCallback = + (messageGroupStore, group) -> this.forceReleaseProcessor.processMessageGroup(group); + store.registerMessageGroupExpiryCallback(expiryCallback); } public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { @@ -746,7 +745,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP 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) { @@ -755,7 +753,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } protected MessageGroup store(Object correlationKey, Message message) { - this.groupIds.add(correlationKey); return this.messageStore.addMessageToGroup(correlationKey, message); } @@ -949,9 +946,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override public Object processMessageGroup(MessageGroup group) { - if (AbstractCorrelatingMessageHandler.this.groupIds.contains(group.getGroupId())) { - forceComplete(group); - } + forceComplete(group); return null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 75582be2fa..8b2a2e4520 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -42,7 +42,7 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG protected final Log logger = LogFactory.getLog(getClass()); - private final Collection expiryCallbacks = new LinkedHashSet(); + private final Collection expiryCallbacks = new LinkedHashSet<>(); private final MessageGroupFactory persistentMessageGroupFactory = new SimpleMessageGroupFactory(SimpleMessageGroupFactory.GroupType.PERSISTENT); @@ -72,13 +72,10 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG /** * Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply * be registered with the store using {@link #registerMessageGroupExpiryCallback(MessageGroupCallback)}. - * * @param expiryCallbacks the expiry callbacks to add */ public void setExpiryCallbacks(Collection expiryCallbacks) { - for (MessageGroupCallback callback : expiryCallbacks) { - registerMessageGroupExpiryCallback(callback); - } + expiryCallbacks.forEach(this::registerMessageGroupExpiryCallback); } public boolean isTimeoutOnIdle() { @@ -110,6 +107,17 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG @Override public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { + if (callback instanceof UniqueExpiryCallback) { + boolean uniqueExpiryCallbackPresent = + this.expiryCallbacks.stream() + .anyMatch(UniqueExpiryCallback.class::isInstance); + + if (!uniqueExpiryCallbackPresent && this.logger.isErrorEnabled()) { + this.logger.error("Only one instance of 'UniqueExpiryCallback' can be registered in the " + + "'MessageGroupStore'. Use a separate 'MessageGroupStore' for each aggregator/resequencer."); + } + } + this.expiryCallbacks.add(callback); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/UniqueExpiryCallback.java b/spring-integration-core/src/main/java/org/springframework/integration/store/UniqueExpiryCallback.java new file mode 100644 index 0000000000..ee9f91dd57 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/UniqueExpiryCallback.java @@ -0,0 +1,33 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.store; + +/** + * A marker interface extension of the {@link MessageGroupStore.MessageGroupCallback} + * for components which should be registered in the {@link MessageGroupStore} only once. + * The {@link MessageGroupStore} implementation ensures that only once instance of this + * class is present in the expire callbacks. + * + * @author Meherzad Lahewala + * @author Artme Bilan + * + * @since 5.0.10 + */ +@FunctionalInterface +public interface UniqueExpiryCallback extends MessageGroupStore.MessageGroupCallback { + +} 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 2ed4190970..c3c4501977 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 @@ -425,6 +425,7 @@ public class AbstractCorrelatingMessageHandlerTests { } @Test + @Ignore("Until 5.2 with new 'owner' feature on groups") public void testDontReapMessageOfOtherHandler() { MessageGroupStore groupStore = new SimpleMessageStore(); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java index 64cd48b812..809a0546c3 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java @@ -17,12 +17,12 @@ package org.springframework.integration.mongodb.store; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.Iterator; -import java.util.LinkedHashSet; import java.util.List; import java.util.UUID; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; import org.springframework.data.domain.Sort; import org.springframework.data.mongodb.MongoDbFactory; @@ -32,12 +32,10 @@ import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import org.springframework.integration.store.MessageGroup; -import org.springframework.integration.store.MessageGroupMetadata; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.jmx.export.annotation.ManagedAttribute; -import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -59,10 +57,6 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb public final static String DEFAULT_COLLECTION_NAME = "configurableStoreMessages"; - private final Collection expiryCallbacks = new LinkedHashSet<>(); - - private volatile boolean timeoutOnIdle; - public ConfigurableMongoDbMessageStore(MongoTemplate mongoTemplate) { this(mongoTemplate, DEFAULT_COLLECTION_NAME); @@ -86,35 +80,10 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb public ConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, String collectionName) { + super(mongoDbFactory, mappingMongoConverter, collectionName); } - /** - * Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply - * be registered with the store using {@link #registerMessageGroupExpiryCallback(MessageGroupCallback)}. - * @param expiryCallbacks the expiry callbacks to add - */ - public void setExpiryCallbacks(Collection expiryCallbacks) { - for (MessageGroupCallback callback : expiryCallbacks) { - registerMessageGroupExpiryCallback(callback); - } - } - - public boolean isTimeoutOnIdle() { - return this.timeoutOnIdle; - } - - /** - * Allows you to override the rule for the timeout calculation. Typical timeout is based from the time - * the {@link MessageGroup} was created. If you want the timeout to be based on the time - * the {@link MessageGroup} was idling (e.g., inactive from the last update) invoke this method with 'true'. - * Default is 'false'. - * @param timeoutOnIdle The boolean. - */ - public void setTimeoutOnIdle(boolean timeoutOnIdle) { - this.timeoutOnIdle = timeoutOnIdle; - } - @Override public Message addMessage(Message message) { Assert.notNull(message, "'message' must not be null"); @@ -205,7 +174,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messages, "'messageToRemove' must not be null"); - Collection ids = new ArrayList(); + Collection ids = new ArrayList<>(); for (Message messageToRemove : messages) { ids.add(messageToRemove.getHeaders().getId()); if (ids.size() >= getRemoveBatchSize()) { @@ -225,11 +194,6 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb this.mongoTemplate.remove(query, this.collectionName); } - @Override - public void removeMessagesFromGroup(Object groupId, Message... messages) { - removeMessagesFromGroup(groupId, Arrays.asList(messages)); - } - @Override public Message pollMessageFromGroup(final Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); @@ -247,52 +211,24 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { - this.updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.LAST_RELEASED_SEQUENCE, sequenceNumber)); + updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.LAST_RELEASED_SEQUENCE, sequenceNumber)); } @Override public void completeGroup(Object groupId) { - this.updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.COMPLETE, true)); + updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.COMPLETE, true)); } @Override public Iterator iterator() { - List messageGroups = new ArrayList(); - Query query = Query.query(Criteria.where(MessageDocumentFields.GROUP_ID).exists(true)); Iterable groupIds = mongoTemplate.getCollection(collectionName) .distinct(MessageDocumentFields.GROUP_ID, query.getQueryObject(), String.class); - for (Object groupId : groupIds) { - messageGroups.add(getMessageGroup(groupId)); - } + return StreamSupport.stream(groupIds.spliterator(), false) + .map(this::getMessageGroup) + .iterator(); - return messageGroups.iterator(); - } - - @Override - public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { - this.expiryCallbacks.add(callback); - } - - @Override - @ManagedOperation - public int expireMessageGroups(long timeout) { - int count = 0; - long threshold = System.currentTimeMillis() - timeout; - for (MessageGroup group : this) { - - long timestamp = group.getTimestamp(); - if (this.isTimeoutOnIdle() && group.getLastModified() > 0) { - timestamp = group.getLastModified(); - } - - if (timestamp <= threshold) { - count++; - expire(group); - } - } - return count; } @Override @@ -315,11 +251,6 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb .size(); } - @Override - public MessageGroupMetadata getGroupMetadata(Object groupId) { - throw new UnsupportedOperationException("Not yet implemented for this store"); - } - @Override public Message getOneMessageFromGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); @@ -338,36 +269,12 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb Assert.notNull(groupId, "'groupId' must not be null"); Query query = groupOrderQuery(groupId); List documents = this.mongoTemplate.find(query, MessageDocument.class, this.collectionName); - List> messages = new ArrayList>(); - for (MessageDocument document : documents) { - messages.add(document.getMessage()); - } - return messages; + return documents.stream() + .map(MessageDocument::getMessage) + .collect(Collectors.toList()); } - private void expire(MessageGroup group) { - - RuntimeException exception = null; - - for (MessageGroupCallback callback : this.expiryCallbacks) { - try { - callback.execute(this, group); - } - catch (RuntimeException e) { - if (exception == null) { - exception = e; - } - logger.error("Exception in expiry callback", e); - } - } - - if (exception != null) { - throw exception; - } - } - - private void updateGroup(Object groupId, Update update) { this.mongoTemplate.updateFirst(groupOrderQuery(groupId), update, this.collectionName); } diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 04aabe8a7b..82c31756f3 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -812,6 +812,13 @@ The callback has direct access to the store and the message group so that it can The `MessageGroupStore` maintains a list of these callbacks, which it applies, on demand, to all messages whose timestamps are earlier than a time supplied as a parameter (see the `registerMessageGroupExpiryCallback(..)` and `expireMessageGroups(..)` methods, described earlier). For more detail, see <>. +IMPORTANT: It is important not to use the same `MessageGroupStore` instance in different aggregator components, when you intend to rely on the `expireMessageGroups` functionality. +Every `AbstractCorrelatingMessageHandler` registers its own `MessageGroupCallback` based on the `forceComplete()` callback. +This way each group for expiration may be completed or discarded by the wrong aggregator. +Starting with version 5.0.10, a `UniqueExpiryCallback` is used from the `AbstractCorrelatingMessageHandler` for the registration callback in the `MessageGroupStore`. +The `MessageGroupStore`, in turn, checks for presence an instance of this class and logs an error with an appropriate message if one is already present in the callbacks set. +This way the Framework disallows usage of the `MessageGroupStore` instance in different aggregators/resequencers to avoid the mentioned side effect of expiration the groups not created by the particular correlation handler. + You can call the `expireMessageGroups` method with a timeout value. Any message older than the current time minus this value is expired and has the callbacks applied. Thus, it is the user of the store that defines what is meant by message group "`expiry`".