Skip to content

Commit f4212d8

Browse files
authored
GH-8685: Re-fetch group after setting condition (#8686)
* GH-8685: Re-fetch group after setting condition Fixes #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
1 parent 4e464e9 commit f4212d8

File tree

3 files changed

+39
-7
lines changed

3 files changed

+39
-7
lines changed

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

+13-6
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2022 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -566,7 +566,7 @@ private boolean processMessageForGroup(Message<?> message, Object correlationKey
566566
this.logger.trace(() -> "Adding message to group [ " + messageGroupToLog + "]");
567567
messageGroup = store(correlationKey, message);
568568

569-
setGroupConditionIfAny(message, messageGroup);
569+
messageGroup = setGroupConditionIfAny(message, messageGroup);
570570

571571
if (this.releaseStrategy.canRelease(messageGroup)) {
572572
Collection<Message<?>> completedMessages = null;
@@ -605,12 +605,19 @@ private void cancelScheduledFutureIfAny(Object correlationKey, UUID groupIdUuid,
605605
}
606606
}
607607

608-
private void setGroupConditionIfAny(Message<?> message, MessageGroup messageGroup) {
608+
private MessageGroup setGroupConditionIfAny(Message<?> message, MessageGroup messageGroup) {
609+
MessageGroup messageGroupToUse = messageGroup;
610+
609611
if (this.groupConditionSupplier != null) {
610-
String condition = this.groupConditionSupplier.apply(message, messageGroup.getCondition());
611-
this.messageStore.setGroupCondition(messageGroup.getGroupId(), condition);
612-
messageGroup.setCondition(condition);
612+
String condition = this.groupConditionSupplier.apply(message, messageGroupToUse.getCondition());
613+
this.messageStore.setGroupCondition(messageGroupToUse.getGroupId(), condition);
614+
messageGroupToUse = this.messageStore.getMessageGroup(messageGroupToUse.getGroupId());
615+
if (this.sequenceAware) {
616+
messageGroupToUse = new SequenceAwareMessageGroup(messageGroupToUse);
617+
}
613618
}
619+
620+
return messageGroupToUse;
614621
}
615622

616623
protected boolean isExpireGroupsUponCompletion() {

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java

+24-1
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2022 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -65,6 +65,29 @@ void testWithAggregatorWithShutdown() {
6565
super.testWithAggregatorWithShutdown("mongo-aggregator-configurable-config.xml");
6666
}
6767

68+
@Test
69+
void groupIsForceReleaseAfterTimeoutWhenGroupConditionIsSet() {
70+
try (var context = new ClassPathXmlApplicationContext("mongo-aggregator-configurable-config.xml", getClass())) {
71+
MessageChannel input = context.getBean("inputChannel", MessageChannel.class);
72+
QueueChannel output = context.getBean("outputChannel", QueueChannel.class);
73+
74+
Message<?> message = MessageBuilder.withPayload("test")
75+
.setSequenceNumber(1)
76+
.setSequenceSize(10)
77+
.setCorrelationId("test")
78+
.build();
79+
80+
input.send(message);
81+
82+
Message<?> receive = output.receive(10_000);
83+
84+
assertThat(receive)
85+
.extracting("payload")
86+
.asList()
87+
.hasSize(1);
88+
}
89+
}
90+
6891
@Test
6992
@Disabled("The performance test. Enough slow. Also needs the release strategy changed to size() == 1000")
7093
void messageGroupStoreLazyLoadPerformance() {

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml

+2
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111

1212
<int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="mongoStore"
1313
release-strategy="releaseStrategy"
14+
group-timeout="500"
15+
send-partial-result-on-expiry="true"
1416
group-condition-supplier="conditionSupplier"/>
1517

1618
<util:constant id="releaseStrategy"

0 commit comments

Comments
 (0)