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

INT-2937 Fix Race Condition in MGS Reaper

Browse Source 
Groups changed within the same millisecond that they
were selected for reaping can cause aggregators to
emit duplicate messages.

Use the re-fetched group (used to see if the timetamp
changed) instead of the group passed in to forceClose
(the group might have changed since it was selected as
eligible for reaping).

We still retain the last modified check because it
might have been expired using timeoutOnIdle.
This commit is contained in:
Gary Russell
2013-02-17 14:46:12 -05:00
committed by Mark Fisher
parent 69fcf5b16a
commit eb07178381
2 changed files with 46 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -268,20 +268,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
lock.lockInterruptibly();
try {
/*
* Need to verify the group hasn't changed while we were waiting on
* its lock. We have to re-fetch the group for this. A possible
* future improvement would be to add MessageGroupStore.getLastModified(groupId).
* Refetch the group because it might have changed while we were waiting on
* its lock. If the last modified timestamp changed, defer the completion
* because the selection condition may have changed such that the group
* would no longer be eligible.
*/
MessageGroup messageGroupNow = this.messageStore.getMessageGroup(
MessageGroup groupNow = this.messageStore.getMessageGroup(
group.getGroupId());
long lastModifiedNow = messageGroupNow.getLastModified();
long lastModifiedNow = groupNow.getLastModified();
if (group.getLastModified() == lastModifiedNow) {
if (group.size() > 0) {
if (releaseStrategy.canRelease(group)) {
this.completeGroup(correlationKey, group);
if (groupNow.size() > 0) {
if (releaseStrategy.canRelease(groupNow)) {
this.completeGroup(correlationKey, groupNow);
}
else {
this.expireGroup(correlationKey, group);
this.expireGroup(correlationKey, groupNow);
}
}
else {
@@ -292,14 +293,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
*/
removeGroup = lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
if (removeGroup && logger.isDebugEnabled()) {
logger.debug("Removing empty group: " + group.getGroupId());
logger.debug("Removing empty group: " + correlationKey);
}
}
}
else {
removeGroup = false;
if (logger.isDebugEnabled()) {
logger.debug("Group expiry candidate (" + group.getGroupId() +
logger.debug("Group expiry candidate (" + correlationKey +
") has changed - it may be reconsidered for a future expiration");
}
}

34
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
View File

@@ -16,9 +16,11 @@
package org.springframework.integration.aggregator;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -28,9 +30,11 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
@@ -249,4 +253,34 @@ public class AbstractCorrelatingMessageHandlerTests {
assertEquals(0, TestUtils.getPropertyValue(handler, "messageStore.groupIdToMessageGroup", Map.class).size());
}
@Test
public void testReapWithChangeInSameMillisecond() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
public boolean canRelease(MessageGroup group) {
return true;
}
});
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class);
forceComplete.setAccessible(true);
mgs.addMessageToGroup("foo", new GenericMessage<String>("foo"));
GenericMessage<String> secondMessage = new GenericMessage<String>("bar");
mgs.addMessageToGroup("foo", secondMessage);
MessageGroup group = mgs.getMessageGroup("foo");
// remove a message
mgs.removeMessageFromGroup("foo", secondMessage);
// force lastModified to be the same
MessageGroup groupNow = mgs.getMessageGroup("foo");
new DirectFieldAccessor(group).setPropertyValue("lastModified", groupNow.getLastModified());
forceComplete.invoke(handler, group);
Message<?> message = outputChannel.receive(0);
assertNotNull(message);
Collection<?> payload = (Collection<?>) message.getPayload();
assertEquals(1, payload.size());
}
}