From 638276ff231fcffe0337d4359b64e8941ee710e3 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 | 25 ++++++++++++++++++- .../mongo-aggregator-configurable-config.xml | 2 ++ 3 files changed, 39 insertions(+), 7 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 ac7324918a..198058db7d 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. @@ -566,7 +566,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; @@ -605,12 +605,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 62e2453e5f..d4c74e4707 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 @@ -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. @@ -65,6 +65,29 @@ class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGr super.testWithAggregatorWithShutdown("mongo-aggregator-configurable-config.xml"); } + @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); + } + } + @Test @Disabled("The performance test. Enough slow. Also needs the release strategy changed to size() == 1000") void messageGroupStoreLazyLoadPerformance() { 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 @@