From 34683bd5603ce324fd4c79da3d8a8e8efede846c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 24 Nov 2013 17:28:33 -0500 Subject: [PATCH] INT-3216 MGS Reaper Improvements 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 --- .../AbstractCorrelatingMessageHandler.java | 35 ++++-- ...bstractCorrelatingMessageHandlerTests.java | 118 +++++++++++++++++- 2 files changed, 144 insertions(+), 9 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index d6dacb8e9a..e8b2abe747 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index ecc647d3bd..c1f3a2d929 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -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("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("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("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)); + } + }