Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-8685: Re-fetch group after setting condition (#8686)

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2023-07-24 09:10:44 -04:00
committed by Gary Russell
parent 035b838ec0
commit 11aca689c3
3 changed files with 39 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Message<?>> 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() {