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

Author: artembilan

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 @@ private boolean processMessageForGroup(Message<?> message, Object correlationKey
 			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;
@@ -604,12 +604,19 @@ private void cancelScheduledFutureIfAny(Object correlationKey, UUID groupIdUuid,
 		}
 	}
 
-	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() {

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

@@ -81,6 +81,29 @@ public void messageGroupStoreLazyLoadPerformance() {
 		//		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());

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

@@ -11,6 +11,8 @@
 
 	<int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="mongoStore"
 					release-strategy="releaseStrategy"
+					group-timeout="500"
+					send-partial-result-on-expiry="true"
 					group-condition-supplier="conditionSupplier"/>
 
 	<util:constant id="releaseStrategy"