From 913a8058dd6b2a1130e1e23cae458a101e3e4372 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 17 Sep 2012 21:46:40 +0100 Subject: [PATCH] INT-2751 Fix Reaper Race Condition When the reaper runs, and finds a group that is in the process of being completed, the reaper blocks on the lock, but when the lock is released, goes ahead and reaps the group. Now, after obtaining the lock, the reaper checks to see if the modified date changed and, if so, aborts the reap of the group this time around. Required adding lastModified accounting for SimpleMessageGroup (accounting already exists for other message groups). Test case reproduced the problem, demonstrating duplicate output messages. After the fix, it shows the reap was aborted in the debug log. INT-2751 Polishing PR Comments --- .../AbstractCorrelatingMessageHandler.java | 28 +++- .../integration/store/SimpleMessageGroup.java | 2 + .../integration/store/SimpleMessageStore.java | 8 +- ...bstractCorrelatingMessageHandlerTests.java | 153 ++++++++++++++++++ 4 files changed, 186 insertions(+), 5 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java 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 7746ef21b3..13bac8de46 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 @@ -244,20 +244,40 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH Object correlationKey = group.getGroupId(); Lock lock = this.lockRegistry.obtain(correlationKey); + boolean removeGroup = true; try { lock.lockInterruptibly(); try { if (group.size() > 0) { try { - if (releaseStrategy.canRelease(group)) { - this.completeGroup(correlationKey, group); + /* + * 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). + */ + MessageGroup messageGroupNow = this.messageStore.getMessageGroup( + group.getGroupId()); + long lastModifiedNow = messageGroupNow.getLastModified(); + if (group.getLastModified() == lastModifiedNow) { + if (releaseStrategy.canRelease(group)) { + this.completeGroup(correlationKey, group); + } + else { + this.expireGroup(correlationKey, group); + } } else { - this.expireGroup(correlationKey, group); + removeGroup = false; + if (logger.isDebugEnabled()) { + logger.debug("Group expiry candidate (" + group.getGroupId() + + ") has changed - it may be reconsidered for a future expiration"); + } } } finally { - this.remove(group); + if (removeGroup) { + this.remove(group); + } } return true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index b57078ae29..bb266a1ea3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -27,6 +27,7 @@ import org.springframework.integration.Message; * @author Iwein Fuld * @author Oleg Zhurakousky * @author Dave Syer + * @author Gary Russell * @since 2.0 */ public class SimpleMessageGroup implements MessageGroup { @@ -144,6 +145,7 @@ public class SimpleMessageGroup implements MessageGroup { "groupId=" + groupId + ", messages=" + messages + ", timestamp=" + timestamp + + ", lastModified=" + lastModified + '}'; } } \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index a098e1646f..dc102aa592 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -138,7 +138,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes if (group == null) { return new SimpleMessageGroup(groupId); } - return new SimpleMessageGroup(group); + SimpleMessageGroup simpleMessageGroup = new SimpleMessageGroup(group); + simpleMessageGroup.setLastModified(group.getLastModified()); + return simpleMessageGroup; } public MessageGroup addMessageToGroup(Object groupId, Message message) { @@ -156,6 +158,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes this.groupIdToMessageGroup.putIfAbsent(groupId, group); } group.add(message); + this.groupIdToMessageGroup.get(groupId).setLastModified(System.currentTimeMillis()); return group; } finally { @@ -199,6 +202,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + "can not be located while attempting to remove Message from the MessageGroup"); group.remove(messageToRemove); + group.setLastModified(System.currentTimeMillis()); return group; } finally { @@ -224,6 +228,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); group.setLastReleasedMessageSequenceNumber(sequenceNumber); + group.setLastModified(System.currentTimeMillis()); } finally { lock.unlock(); @@ -244,6 +249,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + "can not be located while attempting to complete the MessageGroup"); group.complete(); + group.setLastModified(System.currentTimeMillis()); } finally { lock.unlock(); 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 new file mode 100644 index 0000000000..5a3d1d46dd --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -0,0 +1,153 @@ +/* + * Copyright 2002-2012 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.aggregator; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +public class AbstractCorrelatingMessageHandlerTests { + + @Test // INT-2751 + public void testReaperDoesntReapAProcessingGroup() throws Exception { + final MessageGroupStore groupStore = new SimpleMessageStore(); + final CountDownLatch waitForSendlatch = new CountDownLatch(1); + final CountDownLatch waitReapStartLatch = new CountDownLatch(1); + final CountDownLatch waitReapCompleteLatch = new CountDownLatch(1); + AbstractCorrelatingMessageHandler handler = new AbstractCorrelatingMessageHandler( + new MessageGroupProcessor() { + + public Object processMessageGroup(MessageGroup group) { + return group; + } + }, groupStore) { + + @Override + protected void afterRelease(MessageGroup group, Collection> completedMessages) { + } + }; + handler.setReleasePartialSequences(true); + + /* + * Runs "reap" when group 'bar' is in completion + */ + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + try { + waitReapStartLatch.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + waitForSendlatch.countDown(); + try { + Thread.sleep(100); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + groupStore.expireMessageGroups(50); + waitReapCompleteLatch.countDown(); + } + }); + + final List> outputMessages = new ArrayList>(); + handler.setOutputChannel(new MessageChannel() { + + /* + * Executes when group 'bar' completes normally + */ + public boolean send(Message message, long timeout) { + outputMessages.add(message); + // wake reaper + waitReapStartLatch.countDown(); + try { + waitForSendlatch.await(10, TimeUnit.SECONDS); + // wait a little longer for reaper to grab groups + Thread.sleep(2000); + // simulate tx commit + groupStore.removeMessageGroup("bar"); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return true; + } + + public boolean send(Message message) { + return this.send(message, 0); + } + }); + handler.setReleaseStrategy(new ReleaseStrategy() { + + public boolean canRelease(MessageGroup group) { + return group.size() == 2; + } + }); + + QueueChannel discards = new QueueChannel(); + handler.setDiscardChannel(discards); + handler.setSendPartialResultOnExpiry(true); + + Message message = MessageBuilder.withPayload("foo") + .setCorrelationId("qux") + .build(); + // partial group that will be reaped + handler.handleMessage(message); + message = MessageBuilder.withPayload("foo") + .setCorrelationId("bar") + .build(); + // full group that should not be reaped + handler.handleMessage(message); + message = MessageBuilder.withPayload("baz") + .setCorrelationId("bar") + .build(); + handler.handleMessage(message); + + assertTrue(waitReapCompleteLatch.await(10, TimeUnit.SECONDS)); + // Before INT-2751 we got bar + bar + qux + assertEquals(2, outputMessages.size()); // bar + qux + // normal release + assertEquals(2, ((MessageGroup) outputMessages.get(0).getPayload()).size()); // 'bar' + // reaper release + assertEquals(1, ((MessageGroup) outputMessages.get(1).getPayload()).size()); // 'qux' + + assertNull(discards.receive(0)); + } + +}