From 6861a16b95d110650b618582c20ebf4b64bbdc39 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 31 Oct 2018 13:33:47 -0400 Subject: [PATCH] INT-4549: Avoiding Aggregator Deadlocks JIRA: https://jira.spring.io/browse/INT-4549 Polishing - PR comments - avoid ThreadLocals Fix javadoc Polishing - discardMessage() Polishing 5.1.1 only * Fix typos and some polishing --- .../AbstractCorrelatingMessageHandler.java | 130 ++++++++++++------ .../config/AggregatorFactoryBean.java | 10 ++ ...stractCorrelatingMessageHandlerParser.java | 3 + .../config/spring-integration-5.1.xsd | 13 ++ .../aggregator/AggregatorTests.java | 128 +++++++++++++++-- .../config/AggregatorParserTests-context.xml | 11 +- .../config/AggregatorParserTests.java | 3 + src/reference/asciidoc/aggregator.adoc | 41 ++++++ src/reference/asciidoc/configuration.adoc | 1 + 9 files changed, 288 insertions(+), 52 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 36daf2b0b4..e09868d05a 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 @@ -118,9 +118,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private String discardChannelName; - private boolean sendPartialResultOnExpiry = false; + private boolean sendPartialResultOnExpiry; - private boolean sequenceAware = false; + private boolean sequenceAware; private LockRegistry lockRegistry = new DefaultLockRegistry(); @@ -144,6 +144,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private boolean popSequence = true; + private boolean releaseLockBeforeSend; + private volatile boolean running; public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, @@ -289,6 +291,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP super.setTaskScheduler(taskScheduler); } + protected boolean isReleaseLockBeforeSend() { + return this.releaseLockBeforeSend; + } + + /** + * Set to true to release the message group lock before sending any output. See + * "Avoiding Deadlocks" in the Aggregator section of the reference manual for more + * information as to why this might be needed. + * @param releaseLockBeforeSend true to release the lock. + * @since 5.1.1 + */ + public void setReleaseLockBeforeSend(boolean releaseLockBeforeSend) { + this.releaseLockBeforeSend = releaseLockBeforeSend; + } + @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; @@ -439,6 +456,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP UUID groupIdUuid = UUIDConverter.getUUID(correlationKey); Lock lock = this.lockRegistry.obtain(groupIdUuid.toString()); + boolean noOutput = true; lock.lockInterruptibly(); try { ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(groupIdUuid); @@ -463,7 +481,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.releaseStrategy.canRelease(messageGroup)) { Collection> completedMessages = null; try { - completedMessages = completeGroup(message, correlationKey, messageGroup); + noOutput = false; + completedMessages = completeGroup(message, correlationKey, messageGroup, lock); } finally { // Possible clean (implementation dependency) up @@ -479,11 +498,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } else { - discardMessage(message); + noOutput = false; + discardMessage(message, lock); } } finally { - lock.unlock(); + if (noOutput || !this.releaseLockBeforeSend) { + lock.unlock(); + } } } @@ -586,6 +608,13 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } + private void discardMessage(Message message, Lock lock) { + if (this.releaseLockBeforeSend) { + lock.unlock(); + } + discardMessage(message); + } + private void discardMessage(Message message) { this.messagingTemplate.send(getDiscardChannel(), message); } @@ -609,11 +638,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } protected void forceComplete(MessageGroup group) { - Object correlationKey = group.getGroupId(); // UUIDConverter is no-op if already converted Lock lock = this.lockRegistry.obtain(UUIDConverter.getUUID(correlationKey).toString()); boolean removeGroup = true; + boolean noOutput = true; try { lock.lockInterruptibly(); try { @@ -653,11 +682,12 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP && group.getTimestamp() == groupNow.getTimestamp()) { if (groupSize > 0) { + noOutput = false; if (this.releaseStrategy.canRelease(groupNow)) { - completeGroup(correlationKey, groupNow); + completeGroup(correlationKey, groupNow, lock); } else { - expireGroup(correlationKey, groupNow); + expireGroup(correlationKey, groupNow, lock); } if (!this.expireGroupsUponTimeout) { afterRelease(groupNow, groupNow.getMessages(), true); @@ -697,11 +727,13 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP finally { try { if (removeGroup) { - this.remove(group); + remove(group); } } finally { - lock.unlock(); + if (noOutput || !this.releaseLockBeforeSend) { + lock.unlock(); + } } } } @@ -727,7 +759,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP return this.messageStore.addMessageToGroup(correlationKey, message); } - protected void expireGroup(Object correlationKey, MessageGroup group) { + protected void expireGroup(Object correlationKey, MessageGroup group, Lock lock) { if (this.logger.isInfoEnabled()) { this.logger.info("Expiring MessageGroup with correlationKey[" + correlationKey + "]"); } @@ -736,7 +768,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.logger.debug("Prematurely releasing partially complete group with key [" + correlationKey + "] to: " + getOutputChannel()); } - completeGroup(correlationKey, group); + completeGroup(correlationKey, group, lock); } else { if (this.logger.isDebugEnabled()) { @@ -744,50 +776,62 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP + correlationKey + "] to: " + (this.discardChannelName != null ? this.discardChannelName : this.discardChannel)); } - for (Message message : group.getMessages()) { - discardMessage(message); + if (this.releaseLockBeforeSend) { + lock.unlock(); } + group.getMessages() + .forEach(this::discardMessage); } if (this.applicationEventPublisher != null) { - this.applicationEventPublisher.publishEvent(new MessageGroupExpiredEvent(this, correlationKey, group - .size(), new Date(group.getLastModified()), new Date(), !this.sendPartialResultOnExpiry)); + this.applicationEventPublisher.publishEvent( + new MessageGroupExpiredEvent(this, correlationKey, group.size(), + new Date(group.getLastModified()), new Date(), !this.sendPartialResultOnExpiry)); } } - protected void completeGroup(Object correlationKey, MessageGroup group) { + protected void completeGroup(Object correlationKey, MessageGroup group, Lock lock) { Message first = null; if (group != null) { first = group.getOne(); } - completeGroup(first, correlationKey, group); + completeGroup(first, correlationKey, group, lock); } @SuppressWarnings("unchecked") - protected Collection> completeGroup(Message message, Object correlationKey, MessageGroup group) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Completing group with correlationKey [" + correlationKey + "]"); - } + protected Collection> completeGroup(Message message, Object correlationKey, MessageGroup group, + Lock lock) { - Object result = this.outputProcessor.processMessageGroup(group); Collection> partialSequence = null; - if (result instanceof Collection) { - verifyResultCollectionConsistsOfMessages((Collection) result); - partialSequence = (Collection>) result; - } - - if (this.popSequence && partialSequence == null && !(result instanceof Message)) { - AbstractIntegrationMessageBuilder messageBuilder; - if (result instanceof AbstractIntegrationMessageBuilder) { - messageBuilder = (AbstractIntegrationMessageBuilder) result; + Object result; + try { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } - else { - messageBuilder = getMessageBuilderFactory() - .withPayload(result) - .copyHeaders(message.getHeaders()); - } - result = messageBuilder.popSequenceDetails(); - } + result = this.outputProcessor.processMessageGroup(group); + if (result instanceof Collection) { + verifyResultCollectionConsistsOfMessages((Collection) result); + partialSequence = (Collection>) result; + } + + if (this.popSequence && partialSequence == null && !(result instanceof Message)) { + AbstractIntegrationMessageBuilder messageBuilder; + if (result instanceof AbstractIntegrationMessageBuilder) { + messageBuilder = (AbstractIntegrationMessageBuilder) result; + } + else { + messageBuilder = getMessageBuilderFactory() + .withPayload(result) + .copyHeaders(message.getHeaders()); + } + result = messageBuilder.popSequenceDetails(); + } + } + finally { + if (this.releaseLockBeforeSend) { + lock.unlock(); + } + } sendOutputs(result, message); return partialSequence; } @@ -870,11 +914,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.size() == 0) { return true; } - Integer messageSequenceNumber = message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, - Integer.class); + Integer messageSequenceNumber = message.getHeaders() + .get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Integer.class); if (messageSequenceNumber != null && messageSequenceNumber > 0) { - Integer messageSequenceSize = message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, - Integer.class); + Integer messageSequenceSize = message.getHeaders() + .get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Integer.class); if (messageSequenceSize == null) { messageSequenceSize = 0; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index f913dbadd1..c411283c2c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -89,6 +89,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe private Boolean popSequence; + private Boolean releaseLockBeforeSend; + public void setProcessorBean(Object processorBean) { this.processorBean = processorBean; } @@ -173,6 +175,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe this.popSequence = popSequence; } + public void setReleaseLockBeforeSend(Boolean releaseLockBeforeSend) { + this.releaseLockBeforeSend = releaseLockBeforeSend; + } + @Override protected AggregatingMessageHandler createHandler() { MessageGroupProcessor outputProcessor; @@ -265,6 +271,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe aggregator.setPopSequence(this.popSequence); } + if (this.releaseLockBeforeSend != null) { + aggregator.setReleaseLockBeforeSend(this.releaseLockBeforeSend); + } + return aggregator; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index 0d0b792111..2d7e9dd93f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -63,6 +63,8 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo private static final String EXPIRE_GROUPS_UPON_TIMEOUT = "expire-groups-upon-timeout"; + private static final String RELEASE_LOCK = "release-lock-before-send"; + protected void doParse(BeanDefinitionBuilder builder, Element element, BeanMetadataElement processor, ParserContext parserContext) { IntegrationNamespaceUtils.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, @@ -97,6 +99,7 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo builder.getRawBeanDefinition(), parserContext, "forceReleaseAdviceChain"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_TIMEOUT); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_LOCK); } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.1.xsd index e775a05cc5..076793afe4 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.1.xsd @@ -3915,6 +3915,19 @@ + + + + Set to true to release the message group lock before sending any + output. See "Avoiding Deadlocks" in the Aggregator section of + the reference manual for more information as to why this might + be needed. + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 53ba5412a7..5014cffa51 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -50,6 +50,9 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroupFactory; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.UUIDConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; @@ -255,9 +258,14 @@ public class AggregatorTests { @Test public void testCompleteGroupWithinTimeout() { QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); + MessageChannel lockCheckingChannel = (m, to) -> { + checkLock(this.aggregator, "ABC", true); + replyChannel.send(m); + return true; + }; + Message message1 = createMessage(3, "ABC", 3, 1, lockCheckingChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, lockCheckingChannel, null); + Message message3 = createMessage(7, "ABC", 3, 3, lockCheckingChannel, null); this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message2); @@ -268,10 +276,61 @@ public class AggregatorTests { assertEquals(reply.getPayload(), 105); } + @Test + public void testCompleteGroupWithinTimeoutUnlockB4Send() { + QueueChannel replyChannel = new QueueChannel(); + MessageChannel lockCheckingChannel = (m, to) -> { + checkLock(this.aggregator, "ABC", false); + replyChannel.send(m); + return true; + }; + Message message1 = createMessage(3, "ABC", 3, 1, lockCheckingChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, lockCheckingChannel, null); + Message message3 = createMessage(7, "ABC", 3, 3, lockCheckingChannel, null); + + this.aggregator.setReleaseLockBeforeSend(true); + this.aggregator.handleMessage(message1); + this.aggregator.handleMessage(message2); + this.aggregator.handleMessage(message3); + + Message reply = replyChannel.receive(10000); + assertNotNull(reply); + assertEquals(reply.getPayload(), 105); + } + @Test public void testShouldNotSendPartialResultOnTimeoutByDefault() { QueueChannel discardChannel = new QueueChannel(); - this.aggregator.setDiscardChannel(discardChannel); + this.aggregator.setDiscardChannel((m, to) -> { + checkLock(this.aggregator, "ABC", true); + discardChannel.send(m); + return true; + }); + QueueChannel replyChannel = new QueueChannel(); + Message message = createMessage(3, "ABC", 2, 1, replyChannel, null); + this.aggregator.handleMessage(message); + this.store.expireMessageGroups(-10000); + Message reply = replyChannel.receive(0); + assertNull("No message should have been sent normally", reply); + Message discardedMessage = discardChannel.receive(1000); + assertNotNull("A message should have been discarded", discardedMessage); + assertEquals(message, discardedMessage); + assertEquals(1, expiryEvents.size()); + assertSame(this.aggregator, expiryEvents.get(0).getSource()); + assertEquals("ABC", this.expiryEvents.get(0).getGroupId()); + assertEquals(1, this.expiryEvents.get(0).getMessageCount()); + assertTrue(this.expiryEvents.get(0).isDiscarded()); + } + + @Test + public void testShouldNotSendPartialResultOnTimeoutByDefaultUnlockB4Send() { + QueueChannel discardChannel = new QueueChannel(); + this.aggregator.setDiscardChannel((m, to) -> { + checkLock(this.aggregator, "ABC", false); + discardChannel.send(m); + return true; + }); + this.aggregator.setReleaseLockBeforeSend(true); QueueChannel replyChannel = new QueueChannel(); Message message = createMessage(3, "ABC", 2, 1, replyChannel, null); this.aggregator.handleMessage(message); @@ -315,10 +374,19 @@ public class AggregatorTests { this.aggregator.setSendPartialResultOnExpiry(true); this.aggregator.setExpireGroupsUponTimeout(false); QueueChannel replyChannel = new QueueChannel(); + MessageChannel lockCheckingChannel = (m, to) -> { + checkLock(this.aggregator, "ABC", true); + replyChannel.send(m); + return true; + }; QueueChannel discardChannel = new QueueChannel(); - this.aggregator.setDiscardChannel(discardChannel); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); + this.aggregator.setDiscardChannel((m, to) -> { + checkLock(this.aggregator, "ABC", true); + discardChannel.send(m); + return true; + }); + Message message1 = createMessage(3, "ABC", 3, 1, lockCheckingChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, lockCheckingChannel, null); this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message2); this.store.expireMessageGroups(-10000); @@ -331,7 +399,46 @@ public class AggregatorTests { assertEquals(2, this.expiryEvents.get(0).getMessageCount()); assertFalse(this.expiryEvents.get(0).isDiscarded()); assertEquals(0, this.store.getMessageGroup("ABC").size()); - Message message3 = createMessage(5, "ABC", 3, 3, replyChannel, null); + Message message3 = createMessage(5, "ABC", 3, 3, lockCheckingChannel, null); + this.aggregator.handleMessage(message3); + assertEquals(0, this.store.getMessageGroup("ABC").size()); + Message discardedMessage = discardChannel.receive(1000); + assertNotNull("A message should have been discarded", discardedMessage); + assertSame(message3, discardedMessage); + } + + @Test + public void testGroupRemainsAfterTimeoutUnlockB4Discard() { + this.aggregator.setSendPartialResultOnExpiry(true); + this.aggregator.setExpireGroupsUponTimeout(false); + this.aggregator.setReleaseLockBeforeSend(true); + QueueChannel replyChannel = new QueueChannel(); + MessageChannel lockCheckingChannel = (m, to) -> { + checkLock(this.aggregator, "ABC", false); + replyChannel.send(m); + return true; + }; + QueueChannel discardChannel = new QueueChannel(); + this.aggregator.setDiscardChannel((m, to) -> { + checkLock(this.aggregator, "ABC", false); + discardChannel.send(m); + return true; + }); + Message message1 = createMessage(3, "ABC", 3, 1, lockCheckingChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, lockCheckingChannel, null); + this.aggregator.handleMessage(message1); + this.aggregator.handleMessage(message2); + this.store.expireMessageGroups(-10000); + Message reply = replyChannel.receive(1000); + assertNotNull("A reply message should have been received", reply); + assertEquals(15, reply.getPayload()); + assertEquals(1, expiryEvents.size()); + assertSame(this.aggregator, expiryEvents.get(0).getSource()); + assertEquals("ABC", this.expiryEvents.get(0).getGroupId()); + assertEquals(2, this.expiryEvents.get(0).getMessageCount()); + assertFalse(this.expiryEvents.get(0).isDiscarded()); + assertEquals(0, this.store.getMessageGroup("ABC").size()); + Message message3 = createMessage(5, "ABC", 3, 3, lockCheckingChannel, null); this.aggregator.handleMessage(message3); assertEquals(0, this.store.getMessageGroup("ABC").size()); Message discardedMessage = discardChannel.receive(1000); @@ -460,6 +567,11 @@ public class AggregatorTests { return builder.build(); } + private void checkLock(AbstractCorrelatingMessageHandler handler, String group, boolean expectedHeld) { + ReentrantLock lock = (ReentrantLock) TestUtils.getPropertyValue(handler, "lockRegistry", LockRegistry.class) + .obtain(UUIDConverter.getUUID(group).toString()); + assertEquals(expectedHeld, lock.isHeldByCurrentThread()); + } private class MultiplyingProcessor implements MessageGroupProcessor { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml index 1d988602c6..8b28fedd5c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml @@ -3,9 +3,17 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans" xmlns:task="http://www.springframework.org/schema/task" + xmlns:context="http://www.springframework.org/schema/context" + xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd - http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd + http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd + http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd"> + + + + @@ -16,6 +24,7 @@ diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 160057c17a..a893087fd1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -99,6 +99,7 @@ public class AggregatorParserTests { Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME); Object handler = context.getBean("aggregatorWithReference.handler"); assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.messageBuilderFactory")); + assertTrue(TestUtils.getPropertyValue(handler, "releaseLockBeforeSend", Boolean.class)); } @Test @@ -115,6 +116,8 @@ public class AggregatorParserTests { assertEquals(3, output.getQueueSize()); output.purge(null); + assertFalse(TestUtils.getPropertyValue(context.getBean("aggregatorWithMGPReference.handler"), + "releaseLockBeforeSend", Boolean.class)); } @Test diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 4a13c660ce..04aabe8a7b 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -279,6 +279,47 @@ A `LockRegistry` is used to obtain a lock for the resolved correlation ID. A `DefaultLockRegistry` is used by default (in-memory). For synchronizing updates across servers where a shared `MessageGroupStore` is being used, you must configure a shared lock registry. +[[aggregator-deadlocks]] +===== Avoiding Deadlocks + +As discussed above, when message groups are mutated (messages added or released) a lock is held. + +Consider the following flow: + +==== +[source] +---- +...->aggregator1-> ... ->aggregator2-> ... +---- +==== + +If there are multiple threads, **and the aggregators share a common lock registry**, it is possible to get a deadlock. +This will cause hung threads and `jstack ` might present a result such as: + +==== +[source] +---- +Found one Java-level deadlock: +============================= +"t2": + waiting for ownable synchronizer 0x000000076c1cbfa0, (a java.util.concurrent.locks.ReentrantLock$NonfairSync), + which is held by "t1" +"t1": + waiting for ownable synchronizer 0x000000076c1ccc00, (a java.util.concurrent.locks.ReentrantLock$NonfairSync), + which is held by "t2" +---- +==== + +There are several ways to avoid this problem: + +* ensure each aggregator has its own lock registry (this can be a shared registry across application instances but two or more aggregators in the flow must each have a distinct registry) +* use an `ExecutorChannel` or `QueueChannel` as the output channel of the aggregator so that the downstream flow runs on a new thread +* starting with version 5.1.1, set the `releaseLockBeforeSend` aggregator property to `true` + +NOTE: This problem can also be caused if, for some reason, the output of a single aggregator is eventually routed back to the same aggregator. +Of course, the first solution above does not apply in this case. + + [[aggregator-java-dsl]] ==== Configuring an Aggregator in Java DSL diff --git a/src/reference/asciidoc/configuration.adoc b/src/reference/asciidoc/configuration.adoc index b11972a88f..49fdf1651c 100644 --- a/src/reference/asciidoc/configuration.adoc +++ b/src/reference/asciidoc/configuration.adoc @@ -286,6 +286,7 @@ spring.integration.channels.maxBroadcastSubscribers=0x7fffffff spring.integration.readOnly.headers= spring.integration.messagingTemplate.throwExceptionOnLateReply=true ---- +==== [[annotations]] === Annotation Support