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
This commit is contained in:
committed by
Gary Russell
parent
e73894bf74
commit
638276ff23
@@ -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() {
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user