From 07a17206bd887e8ff55357cd544a67cd4029615b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 24 Jul 2023 09:10:44 -0400 Subject: [PATCH] GH-8685: Re-fetch group after setting condition (#8686) * GH-8685: Re-fetch group after setting condition Fixes https://github.com/spring-projects/spring-integration/issues/8685 The `AbstractCorrelatingMessageHandler` updates the group metadata in DB not only for provided `condition`, but also a `lastModified` field. A subsequent scheduling for group timeout takes the `lastModified` to compare with the value in the store after re-fetching group in task. This does not reflect reality since adding `condition` modifies the data in DB, but in-memory state remains the same. * Re-fetch a group from the store in the `AbstractCorrelatingMessageHandler.setGroupConditionIfAny()`. * Verify expected behavior via new `ConfigurableMongoDbMessageGroupStoreTests.groupIsForceReleaseAfterTimeoutWhenGroupConditionIsSet()` **Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`** * * Fix Checkstyle violation in the test --- .../AbstractCorrelatingMessageHandler.java | 19 ++++++++++----- ...igurableMongoDbMessageGroupStoreTests.java | 23 +++++++++++++++++++ .../mongo-aggregator-configurable-config.xml | 2 ++ 3 files changed, 38 insertions(+), 6 deletions(-) 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 fa6184a7f3..c4c1dc3236 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -565,7 +565,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.logger.trace(() -> "Adding message to group [ " + messageGroupToLog + "]"); messageGroup = store(correlationKey, message); - setGroupConditionIfAny(message, messageGroup); + messageGroup = setGroupConditionIfAny(message, messageGroup); if (this.releaseStrategy.canRelease(messageGroup)) { Collection> completedMessages = null; @@ -604,12 +604,19 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } - private void setGroupConditionIfAny(Message message, MessageGroup messageGroup) { + private MessageGroup setGroupConditionIfAny(Message message, MessageGroup messageGroup) { + MessageGroup messageGroupToUse = messageGroup; + if (this.groupConditionSupplier != null) { - String condition = this.groupConditionSupplier.apply(message, messageGroup.getCondition()); - this.messageStore.setGroupCondition(messageGroup.getGroupId(), condition); - messageGroup.setCondition(condition); + String condition = this.groupConditionSupplier.apply(message, messageGroupToUse.getCondition()); + this.messageStore.setGroupCondition(messageGroupToUse.getGroupId(), condition); + messageGroupToUse = this.messageStore.getMessageGroup(messageGroupToUse.getGroupId()); + if (this.sequenceAware) { + messageGroupToUse = new SequenceAwareMessageGroup(messageGroupToUse); + } } + + return messageGroupToUse; } protected boolean isExpireGroupsUponCompletion() { diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java index e3c9d3cd38..2a6c6b6f0f 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java @@ -81,6 +81,29 @@ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMe // System. out .println(watch.prettyPrint()); // checkstyle } + @Test + void groupIsForceReleaseAfterTimeoutWhenGroupConditionIsSet() { + try (var context = new ClassPathXmlApplicationContext("mongo-aggregator-configurable-config.xml", getClass())) { + MessageChannel input = context.getBean("inputChannel", MessageChannel.class); + QueueChannel output = context.getBean("outputChannel", QueueChannel.class); + + Message message = MessageBuilder.withPayload("test") + .setSequenceNumber(1) + .setSequenceSize(10) + .setCorrelationId("test") + .build(); + + input.send(message); + + Message receive = output.receive(10_000); + + assertThat(receive) + .extracting("payload") + .asList() + .hasSize(1); + } + } + private void performLazyLoadEagerTest(StopWatch watch, int sequenceSize, boolean lazyLoad) { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("mongo-aggregator-configurable-config.xml", getClass()); diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml index 552b5ec778..4a407d8429 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml @@ -11,6 +11,8 @@