From 6780bbd5c8e27ce1d88829c0b02c873a78f3811c Mon Sep 17 00:00:00 2001 From: Michel Jung Date: Fri, 21 Aug 2020 16:15:11 +0200 Subject: [PATCH] GH-3367: Separate timeouts in BarrierMH Fixes https://github.com/spring-projects/spring-integration/issues/3367 Introduce a `requestTimeout` and `triggerTimeout` for `BarrierMessageHandler` For instance, if an HTTP request sends a message to the barrier, it should time out after 1min if no trigger message is received. If the trigger message then arrives late and the HTTP request is no longer waiting, it shouldn't wait for 1min before discarding the request but do so immediately. --- .../aggregator/BarrierMessageHandler.java | 91 +++++++++++++++---- .../integration/config/xml/BarrierParser.java | 10 +- .../integration/config/spring-integration.xsd | 13 ++- .../config/xml/BarrierParserTests-context.xml | 3 +- .../config/xml/BarrierParserTests.java | 20 ++-- src/reference/asciidoc/barrier.adoc | 21 ++--- src/reference/asciidoc/whats-new.adoc | 2 + 7 files changed, 113 insertions(+), 47 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java index 26093e03b8..fd0ed2475e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -39,7 +39,7 @@ import org.springframework.util.Assert; * the timeout occurs. Only one thread with a particular correlation (result of invoking * the {@link CorrelationStrategy}) can be suspended at a time. If the inbound thread does * not arrive before the trigger thread, the latter is suspended until it does, or the - * timeout occurs. + * timeout occurs. Separate timeouts may be configured for request and trigger messages. *

* The default {@link CorrelationStrategy} is a {@link HeaderAttributeCorrelationStrategy}. *

@@ -47,6 +47,7 @@ import org.springframework.util.Assert; * * @author Gary Russell * @author Artem Bilan + * @author Michel Jung * * @since 4.2 */ @@ -57,7 +58,9 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler private final Map inProcess = new ConcurrentHashMap<>(); - private final long timeout; + private final long requestTimeout; + + private final long triggerTimeout; private final CorrelationStrategy correlationStrategy; @@ -70,48 +73,100 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler /** * Construct an instance with the provided timeout and default correlation and * output strategies. - * @param timeout the timeout in milliseconds. + * @param timeout the timeout in milliseconds for both, request and trigger messages. */ public BarrierMessageHandler(long timeout) { - this(timeout, new DefaultAggregatingMessageGroupProcessor()); + this(timeout, timeout); } /** * Construct an instance with the provided timeout and output processor, and default * correlation strategy. - * @param timeout the timeout in milliseconds. + * @param timeout the timeout in milliseconds for both, request and trigger messages. * @param outputProcessor the output {@link MessageGroupProcessor}. */ public BarrierMessageHandler(long timeout, MessageGroupProcessor outputProcessor) { - this(timeout, outputProcessor, null); + this(timeout, timeout, outputProcessor); } /** * Construct an instance with the provided timeout and correlation strategy, and default * output processor. - * @param timeout the timeout in milliseconds. + * @param timeout the timeout in milliseconds for both, request and trigger messages. * @param correlationStrategy the correlation strategy. */ public BarrierMessageHandler(long timeout, CorrelationStrategy correlationStrategy) { - this(timeout, new DefaultAggregatingMessageGroupProcessor(), correlationStrategy); + this(timeout, timeout, correlationStrategy); } /** * Construct an instance with the provided timeout and output processor, and default * correlation strategy. - * @param timeout the timeout in milliseconds. + * @param timeout the timeout in milliseconds for both, request and trigger messages. * @param outputProcessor the output {@link MessageGroupProcessor}. * @param correlationStrategy the correlation strategy. */ public BarrierMessageHandler(long timeout, MessageGroupProcessor outputProcessor, CorrelationStrategy correlationStrategy) { + this(timeout, timeout, outputProcessor, correlationStrategy); + } + + /** + * Construct an instance with the provided timeouts and default correlation and + * output strategies. + * @param requestTimeout the timeout in milliseconds when waiting for trigger message. + * @param triggerTimeout the timeout in milliseconds when waiting for a request message. + * @since 5.4 + */ + public BarrierMessageHandler(long requestTimeout, long triggerTimeout) { + this(requestTimeout, triggerTimeout, new DefaultAggregatingMessageGroupProcessor()); + } + + /** + * Construct an instance with the provided timeout and output processor, and default + * correlation strategy. + * @param requestTimeout the timeout in milliseconds when waiting for trigger message. + * @param triggerTimeout the timeout in milliseconds when waiting for a request message. + * @param outputProcessor the output {@link MessageGroupProcessor}. + * @since 5.4 + */ + public BarrierMessageHandler(long requestTimeout, long triggerTimeout, MessageGroupProcessor outputProcessor) { + this(requestTimeout, triggerTimeout, outputProcessor, null); + } + + /** + * Construct an instance with the provided timeout and correlation strategy, and default + * output processor. + * @param requestTimeout the timeout in milliseconds when waiting for trigger message. + * @param triggerTimeout the timeout in milliseconds when waiting for a request message. + * @param correlationStrategy the correlation strategy. + * @since 5.4 + */ + public BarrierMessageHandler(long requestTimeout, long triggerTimeout, CorrelationStrategy correlationStrategy) { + this(requestTimeout, triggerTimeout, new DefaultAggregatingMessageGroupProcessor(), correlationStrategy); + } + + /** + * Construct an instance with the provided timeout and output processor, and default + * correlation strategy. + * @param requestTimeout the timeout in milliseconds when waiting for trigger message. + * @param triggerTimeout the timeout in milliseconds when waiting for a request message. + * @param outputProcessor the output {@link MessageGroupProcessor}. + * @param correlationStrategy the correlation strategy. + * @since 5.4 + */ + public BarrierMessageHandler(long requestTimeout, long triggerTimeout, MessageGroupProcessor outputProcessor, + CorrelationStrategy correlationStrategy) { + Assert.notNull(outputProcessor, "'messageGroupProcessor' cannot be null"); this.messageGroupProcessor = outputProcessor; - this.correlationStrategy = (correlationStrategy == null - ? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) - : correlationStrategy); - this.timeout = timeout; + this.correlationStrategy = + correlationStrategy == null + ? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) + : correlationStrategy; + this.requestTimeout = requestTimeout; + this.triggerTimeout = triggerTimeout; } /** @@ -163,12 +218,12 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler } Thread existing = this.inProcess.putIfAbsent(key, Thread.currentThread()); if (existing != null) { - throw new MessagingException(requestMessage, "Correlation key (" - + key + ") is already in use by " + existing.getName()); + throw new MessagingException(requestMessage, + "Correlation key (" + key + ") is already in use by " + existing.getName()); } SynchronousQueue> syncQueue = createOrObtainQueue(key); try { - Message releaseMessage = syncQueue.poll(this.timeout, TimeUnit.MILLISECONDS); + Message releaseMessage = syncQueue.poll(this.requestTimeout, TimeUnit.MILLISECONDS); if (releaseMessage != null) { return processRelease(key, requestMessage, releaseMessage); } @@ -228,7 +283,7 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler } SynchronousQueue> syncQueue = createOrObtainQueue(key); try { - if (!syncQueue.offer(message, this.timeout, TimeUnit.MILLISECONDS)) { + if (!syncQueue.offer(message, this.triggerTimeout, TimeUnit.MILLISECONDS)) { this.logger.error("Suspending thread timed out or did not arrive within timeout for: " + message); this.suspensions.remove(key); MessageChannel messageChannel = getDiscardChannel(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java index c37a006e1d..7a66f83596 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -27,6 +27,7 @@ import org.springframework.util.StringUtils; * Parser for {@code }. * * @author Gary Russell + * @author Artem Bilan * * @since 4.2 */ @@ -34,9 +35,12 @@ public class BarrierParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinitionBuilder handlerBuilder = BeanDefinitionBuilder - .genericBeanDefinition(BarrierMessageHandler.class); + BeanDefinitionBuilder handlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(BarrierMessageHandler.class); handlerBuilder.addConstructorArgValue(element.getAttribute("timeout")); + String triggerTimeout = element.getAttribute("trigger-timeout"); + if (StringUtils.hasText(triggerTimeout)) { + handlerBuilder.addConstructorArgValue(triggerTimeout); + } String processor = element.getAttribute("output-processor"); if (StringUtils.hasText(processor)) { handlerBuilder.addConstructorArgReference(processor); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd index 19645b99cf..aa572b4b25 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd @@ -1736,8 +1736,7 @@ - + @@ -1751,7 +1750,15 @@ - + + + + The time in milliseconds to suspend the trigger thread. + If not provided a 'timeout' is used. + + + + Specify whether the barrier must return a non-null value. This value will be diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml index c100bd14e3..9e46420924 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml @@ -11,7 +11,8 @@ + timeout="10000" + trigger-timeout="5000"> diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java index 54ce5b66c0..30e85ceba5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -18,8 +18,7 @@ package org.springframework.integration.config.xml; import static org.assertj.core.api.Assertions.assertThat; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.aggregator.BarrierMessageHandler; @@ -35,16 +34,16 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 * */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class BarrierParserTests { @@ -71,8 +70,8 @@ public class BarrierParserTests { @Test public void parserTestsWithMessage() { - this.in.send(new GenericMessage("foo")); - this.release.send(new GenericMessage("bar")); + this.in.send(new GenericMessage<>("foo")); + this.release.send(new GenericMessage<>("bar")); Message received = out.receive(10000); assertThat(received).isNotNull(); this.barrier1.stop(); @@ -82,7 +81,8 @@ public class BarrierParserTests { public void parserFieldPopulationTests() { BarrierMessageHandler handler = TestUtils.getPropertyValue(this.barrier1, "handler", BarrierMessageHandler.class); - assertThat(TestUtils.getPropertyValue(handler, "timeout")).isEqualTo(10000L); + assertThat(TestUtils.getPropertyValue(handler, "requestTimeout")).isEqualTo(10000L); + assertThat(TestUtils.getPropertyValue(handler, "triggerTimeout")).isEqualTo(5000L); assertThat(TestUtils.getPropertyValue(handler, "requiresReply", Boolean.class)).isTrue(); assertThat(TestUtils.getPropertyValue(this.barrier2, "handler.correlationStrategy")) .isInstanceOf(HeaderAttributeCorrelationStrategy.class); diff --git a/src/reference/asciidoc/barrier.adoc b/src/reference/asciidoc/barrier.adoc index 88e9e2c98c..f02649b833 100644 --- a/src/reference/asciidoc/barrier.adoc +++ b/src/reference/asciidoc/barrier.adoc @@ -3,30 +3,27 @@ Sometimes, we need to suspend a message flow thread until some other asynchronous event occurs. For example, consider an HTTP request that publishes a message to RabbitMQ. -We might wish to not reply to the user until the RabbitMQ broker has issued an acknowledgment that the message was -received. +We might wish to not reply to the user until the RabbitMQ broker has issued an acknowledgment that the message was received. In version 4.2, Spring Integration introduced the `` component for this purpose. The underlying `MessageHandler` is the `BarrierMessageHandler`. -This class also implements -`MessageTriggerAction`, in which a message passed to the `trigger()` method releases a corresponding thread in the -`handleRequestMessage()` method (if present). +This class also implements `MessageTriggerAction`, in which a message passed to the `trigger()` method releases a corresponding thread in the `handleRequestMessage()` method (if present). The suspended thread and trigger thread are correlated by invoking a `CorrelationStrategy` on the messages. -When a message is sent to the `input-channel`, the thread is suspended for up to `timeout` milliseconds, waiting for -a corresponding trigger message. +When a message is sent to the `input-channel`, the thread is suspended for up to `requestTimeout` milliseconds, waiting for a corresponding trigger message. The default correlation strategy uses the `IntegrationMessageHeaderAccessor.CORRELATION_ID` header. When a trigger message arrives with the same correlation, the thread is released. The message sent to the `output-channel` after release is constructed by using a `MessageGroupProcessor`. -By default, the message is a `Collection` of the two payloads, and the headers are merged by using a -`DefaultAggregatingMessageGroupProcessor`. +By default, the message is a `Collection` of the two payloads, and the headers are merged by using a `DefaultAggregatingMessageGroupProcessor`. -CAUTION: If the `trigger()` method is invoked first (or after the main thread times out), it is suspended for up to `timeout` waiting for the suspending message to arrive. +CAUTION: If the `trigger()` method is invoked first (or after the main thread times out), it is suspended for up to `triggerTimeout` waiting for the suspending message to arrive. If you do not want to suspend the trigger thread, consider handing off to a `TaskExecutor` instead so that its thread is suspended instead. +NOTE: Prior version 5.4, there was only one `timeout` option for both request and trigger messages, but in some use-case it is better to have different timeouts for those actions. +Therefore `requestTimeout` and `triggerTimeout` options have been introduced. + The `requires-reply` property determines the action to take if the suspended thread times out before the trigger message arrives. -By default, it is `false`, which means the endpoint returns `null`, the flow ends, and the thread returns to the -caller. +By default, it is `false`, which means the endpoint returns `null`, the flow ends, and the thread returns to the caller. When `true`, a `ReplyRequiredException` is thrown. You can call the `trigger()` method programmatically (obtain the bean reference by using the name, `barrier.handler` -- where `barrier` is the bean name of the barrier endpoint). diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 6315fc5ee2..ca4cd8bb11 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -58,6 +58,8 @@ See <<./aggregator.adoc#aggregator-expiring-groups, Aggregator Expiring Groups>> The legacy metrics that were replaced by Micrometer meters have been removed. +The <<./barrier.adoc#barrier,Thread Barrier>> has now two separate timeout options: `requestTimeout` and `triggerTimeout`. + [[x5.4-tcp]] === TCP Changes