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

INT-3216 MGS Reaper Improvements

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3216

Previously, After a group has been selected for completion
by the reaper, if the group is otherwise completed before
the reap AND expireGroupsOnCompletion is true AND a new
group with the same ID is created, ALL in the same millisecond,
AND the group creation timestamp is used for the reap criteria,
the new group will be incorrectly reaped.

Also, for a simple MGS, it was not necessary to re-fetch the group
if the group has already been marked complete.

- Defer reaping if the group is already complete (unless it's empty).
- Verify the timestamp hasn't changed (indicating a new group)

Add tests

- Verify no refetch of a complete SMG
- Verify no reap of a complete group (after refetch)
- Verify no reap when the timestamp changed
This commit is contained in:
Gary Russell
2013-11-24 17:28:33 -05:00
parent ee6ad2c256
commit 34683bd560
2 changed files with 144 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,6 +21,7 @@ import java.util.concurrent.locks.Lock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
@@ -126,6 +127,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
public void setMessageStore(MessageGroupStore store) {
this.messageStore = store;
store.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
@Override
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
forceComplete(group);
}
@@ -143,6 +145,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
}
@Override
public void setOutputChannel(MessageChannel outputChannel) {
Assert.notNull(outputChannel, "'outputChannel' must not be null");
this.outputChannel = outputChannel;
@@ -267,17 +270,32 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
try {
lock.lockInterruptibly();
try {
MessageGroup groupNow = group;
/*
* Refetch the group because it might have changed while we were waiting on
* If the group argument is not already complete,
* re-fetch it 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.
* would no longer be eligible. If the timestamp changed, it's a completely new
* group and should not be reaped on this cycle.
*
* If the group argument is already complete, do not re-fetch.
* Note: not all message stores provide a direct reference to its internal
* group so the initial 'isComplete()` will only return true for those stores if
* the group was already complete at the time of its selection as a candidate.
*
* If the group is marked complete, only consider it
* for reaping if it's empty (and both timestamps are unaltered).
*/
MessageGroup groupNow = this.messageStore.getMessageGroup(
group.getGroupId());
if (!group.isComplete()) {
groupNow = this.messageStore.getMessageGroup(correlationKey);
}
long lastModifiedNow = groupNow.getLastModified();
if (group.getLastModified() == lastModifiedNow) {
if (groupNow.size() > 0) {
int groupSize = groupNow.size();
if ((!groupNow.isComplete() || groupSize == 0)
&& group.getLastModified() == lastModifiedNow
&& group.getTimestamp() == groupNow.getTimestamp()) {
if (groupSize > 0) {
if (releaseStrategy.canRelease(groupNow)) {
this.completeGroup(correlationKey, groupNow);
}
@@ -305,7 +323,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
}
}
}
finally {
finally {
if (removeGroup) {
this.remove(group);
}
@@ -346,7 +364,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
+ correlationKey + "] to: " + outputChannel);
}
completeGroup(correlationKey, group);
} else {
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Discarding messages of partially complete group with key ["
+ correlationKey + "] to: " + discardChannel);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -16,9 +16,13 @@
package org.springframework.integration.aggregator;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.lang.reflect.Method;
import java.util.ArrayList;
@@ -30,6 +34,7 @@ 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;
@@ -57,6 +62,7 @@ public class AbstractCorrelatingMessageHandlerTests {
AbstractCorrelatingMessageHandler handler = new AbstractCorrelatingMessageHandler(
new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup group) {
return group;
}
@@ -73,6 +79,7 @@ public class AbstractCorrelatingMessageHandlerTests {
*/
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
waitReapStartLatch.await(10, TimeUnit.SECONDS);
@@ -98,6 +105,7 @@ public class AbstractCorrelatingMessageHandlerTests {
/*
* Executes when group 'bar' completes normally
*/
@Override
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
// wake reaper
@@ -115,12 +123,14 @@ public class AbstractCorrelatingMessageHandlerTests {
return true;
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return group.size() == 2;
}
@@ -162,6 +172,7 @@ public class AbstractCorrelatingMessageHandlerTests {
AggregatingMessageHandler handler = new AggregatingMessageHandler(
new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup group) {
return group;
}
@@ -174,17 +185,20 @@ public class AbstractCorrelatingMessageHandlerTests {
/*
* Executes when group 'bar' completes normally
*/
@Override
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
return true;
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return group.size() == 1;
}
@@ -208,6 +222,7 @@ public class AbstractCorrelatingMessageHandlerTests {
AggregatingMessageHandler handler = new AggregatingMessageHandler(
new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup group) {
return group;
}
@@ -220,17 +235,20 @@ public class AbstractCorrelatingMessageHandlerTests {
/*
* Executes when group 'bar' completes normally
*/
@Override
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
return true;
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return group.size() == 1;
}
@@ -258,6 +276,7 @@ public class AbstractCorrelatingMessageHandlerTests {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
@@ -283,4 +302,101 @@ public class AbstractCorrelatingMessageHandlerTests {
assertEquals(1, payload.size());
}
@Test /* INT-3216 */
public void testDontReapIfAlreadyComplete() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
mgs.addMessageToGroup("foo", new GenericMessage<String>("foo"));
mgs.completeGroup("foo");
mgs = spy(mgs);
new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs);
Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class);
forceComplete.setAccessible(true);
MessageGroup group = (MessageGroup) TestUtils.getPropertyValue(mgs, "groupIdToMessageGroup", Map.class).get("foo");
assertTrue(group.isComplete());
forceComplete.invoke(handler, group);
verify(mgs, never()).getMessageGroup("foo");
assertNull(outputChannel.receive(0));
}
/*
* INT-3216 - Verifies the complete early exit is taken after a refresh.
*/
@Test
public void testDontReapIfAlreadyCompleteAfterRefetch() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
mgs.addMessageToGroup("foo", new GenericMessage<String>("foo"));
MessageGroup group = mgs.getMessageGroup("foo");
mgs.completeGroup("foo");
mgs = spy(mgs);
new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs);
Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class);
forceComplete.setAccessible(true);
MessageGroup groupInStore = (MessageGroup) TestUtils.getPropertyValue(mgs, "groupIdToMessageGroup", Map.class).get("foo");
assertTrue(groupInStore.isComplete());
assertFalse(group.isComplete());
new DirectFieldAccessor(group).setPropertyValue("lastModified", groupInStore.getLastModified());
forceComplete.invoke(handler, group);
verify(mgs).getMessageGroup("foo");
assertNull(outputChannel.receive(0));
}
/*
* INT-3216 - Verifies we don't complete if it's a completely new group (different timestamp).
*/
@Test
public void testDontReapIfNewGroupFoundDuringRefetch() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
mgs.addMessageToGroup("foo", new GenericMessage<String>("foo"));
MessageGroup group = mgs.getMessageGroup("foo");
mgs = spy(mgs);
new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs);
Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class);
forceComplete.setAccessible(true);
MessageGroup groupInStore = (MessageGroup) TestUtils.getPropertyValue(mgs, "groupIdToMessageGroup", Map.class).get("foo");
assertFalse(groupInStore.isComplete());
assertFalse(group.isComplete());
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(group);
directFieldAccessor.setPropertyValue("lastModified", groupInStore.getLastModified());
directFieldAccessor.setPropertyValue("timestamp", groupInStore.getTimestamp() - 1);
forceComplete.invoke(handler, group);
verify(mgs).getMessageGroup("foo");
assertNull(outputChannel.receive(0));
}
}