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

Merge pull request #751 from garyrussell/INT-2937

Browse Source 
* INT-2937: Fix Race Condition in MGS Reaper
This commit is contained in:
Mark Fisher
2013-02-25 14:40:08 -05:00
parent 69fcf5b16a eb07178381
commit 9efb70f24e
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());
}
}