From a7843af045afcee48eae519c629d4bae8cbd34b0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 24 Mar 2021 11:12:03 -0400 Subject: [PATCH] GH-3521: Delayer: schedule release task with TX (#3525) * GH-3521: Delayer: schedule release task with TX Fixes https://github.com/spring-projects/spring-integration/issues/3521 There is a race condition when transactional `MessageStore` is used for `DelayHandler`, so the message is not visible for reads until after TX is committed, but a scheduled release task may be already ready after delay * Register a `TransactionSynchronization` with scheduling a releasing task when TX is committed **Cherry-pick to `5.4.x` & `5.3.x`** * Fix language in delayer.adoc Co-authored-by: Gary Russell Co-authored-by: Gary Russell --- .../integration/handler/DelayHandler.java | 21 +- .../config/xml/DelayerUsageTests.java | 57 +++--- .../integration/dsl/KotlinDslTests.kt | 189 +++++++++--------- src/reference/asciidoc/delayer.adoc | 75 ++++--- 4 files changed, 189 insertions(+), 153 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 75349a4035..cc5afd338c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -51,6 +51,8 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.support.ErrorMessage; import org.springframework.scheduling.TaskScheduler; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -438,7 +440,24 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement }; } - getTaskScheduler().schedule(releaseTask, new Date(messageWrapper.getRequestDate() + delay)); + Date startTime = new Date(messageWrapper.getRequestDate() + delay); + + if (TransactionSynchronizationManager.isSynchronizationActive() && + TransactionSynchronizationManager.isActualTransactionActive()) { + + TransactionSynchronizationManager.registerSynchronization( + new TransactionSynchronization() { + + @Override + public void afterCommit() { + getTaskScheduler().schedule(releaseTask, startTime); + } + + }); + } + else { + getTaskScheduler().schedule(releaseTask, startTime); + } } private Message getMessageById(UUID messageId) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java index 0a2d6cd145..d60ab204cc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java @@ -18,27 +18,28 @@ package org.springframework.integration.config.xml; import static org.assertj.core.api.Assertions.assertThat; -import org.junit.Ignore; -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.beans.factory.annotation.Qualifier; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.transaction.PseudoTransactionManager; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.support.TransactionTemplate; /** * @author Oleg Zhurakousky * @author Artem Bilan + * * @since 1.0.3 */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration +@SpringJUnitConfig +@DirtiesContext public class DelayerUsageTests { @Autowired @@ -68,11 +69,17 @@ public class DelayerUsageTests { @Test - public void testDelayWithDefaultScheduler() { + public void testDelayWithDefaultSchedulerAndTransactionSynchronization() { long start = System.currentTimeMillis(); - inputA.send(new GenericMessage("Hello")); + + new TransactionTemplate(new PseudoTransactionManager()) + .execute(status -> { + inputA.send(new GenericMessage<>("Hello")); + return null; + }); + assertThat(outputA.receive(10000)).isNotNull(); - assertThat((System.currentTimeMillis() - start) >= 1000).isTrue(); + assertThat(System.currentTimeMillis() - start).isGreaterThanOrEqualTo(1000); } @Test @@ -83,20 +90,19 @@ public class DelayerUsageTests { long start = System.currentTimeMillis(); inputA.send(builder.build()); assertThat(outputA.receive(10000)).isNotNull(); - assertThat((System.currentTimeMillis() - start) >= 2000).isTrue(); + assertThat(System.currentTimeMillis() - start).isGreaterThanOrEqualTo(2000); } @Test - @Ignore("Enough wonky test based on the timeout and hardware") public void testDelayWithCustomScheduler() { long start = System.currentTimeMillis(); - inputB.send(new GenericMessage("1")); - inputB.send(new GenericMessage("2")); - inputB.send(new GenericMessage("3")); - inputB.send(new GenericMessage("4")); - inputB.send(new GenericMessage("5")); - inputB.send(new GenericMessage("6")); - inputB.send(new GenericMessage("7")); + inputB.send(new GenericMessage<>("1")); + inputB.send(new GenericMessage<>("2")); + inputB.send(new GenericMessage<>("3")); + inputB.send(new GenericMessage<>("4")); + inputB.send(new GenericMessage<>("5")); + inputB.send(new GenericMessage<>("6")); + inputB.send(new GenericMessage<>("7")); assertThat(outputB1.receive(10000)).isNotNull(); assertThat(outputB1.receive(10000)).isNotNull(); assertThat(outputB1.receive(10000)).isNotNull(); @@ -108,27 +114,26 @@ public class DelayerUsageTests { // must execute under 3 seconds, since threadPool is set too 5. // first batch is 5 concurrent invocations on SA, then 2 more // elapsed time for the whole execution should be a bit over 2 seconds depending on the hardware - assertThat(((System.currentTimeMillis() - start) >= 1000) && ((System.currentTimeMillis() - start) < 3000)) - .isTrue(); + assertThat(System.currentTimeMillis() - start).isBetween(1000L, 3000L); } - @Test //INT-1132 + @Test public void testDelayerInsideChain() { long start = System.currentTimeMillis(); - delayerInsideChain.send(new GenericMessage("Hello")); + delayerInsideChain.send(new GenericMessage<>("Hello")); Message message = outputA.receive(10000); assertThat(message).isNotNull(); - assertThat((System.currentTimeMillis() - start) >= 1000).isTrue(); + assertThat(System.currentTimeMillis() - start).isGreaterThanOrEqualTo(1000); assertThat(message.getPayload()).isEqualTo("hello"); } @Test public void testInt2243DelayerExpression() { long start = System.currentTimeMillis(); - this.inputC.send(new GenericMessage("test")); + this.inputC.send(new GenericMessage<>("test")); Message message = this.outputC.receive(10000); assertThat(message).isNotNull(); - assertThat((System.currentTimeMillis() - start) >= 1000).isTrue(); + assertThat(System.currentTimeMillis() - start).isGreaterThanOrEqualTo(1000); assertThat(message.getPayload()).isEqualTo("test"); } diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt index 10248544c4..fadf915e39 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt @@ -18,8 +18,6 @@ package org.springframework.integration.dsl import assertk.assertThat import assertk.assertions.* -import org.apache.commons.logging.Log -import org.apache.commons.logging.LogFactory import org.junit.jupiter.api.Test import org.springframework.beans.factory.BeanFactory import org.springframework.beans.factory.annotation.Autowired @@ -51,6 +49,7 @@ import java.util.* import java.util.concurrent.atomic.AtomicReference import java.util.function.Function + /** * @author Artem Bilan */ @@ -74,16 +73,16 @@ class KotlinDslTests { val replyChannel = QueueChannel() val date = Date() val testMessage = - MessageBuilder.withPayload("{\"name\": \"Test\",\"date\": " + date.time + "}") - .setHeader(MessageHeaders.CONTENT_TYPE, "application/json") - .setReplyChannel(replyChannel) - .build() + MessageBuilder.withPayload("{\"name\": \"Test\",\"date\": " + date.time + "}") + .setHeader(MessageHeaders.CONTENT_TYPE, "application/json") + .setReplyChannel(replyChannel) + .build() this.convertFlowInput.send(testMessage) assertThat(replyChannel.receive(10000)?.payload) - .isNotNull() - .isInstanceOf(TestPojo::class.java) - .isEqualTo(TestPojo("Test", date)) + .isNotNull() + .isInstanceOf(TestPojo::class.java) + .isEqualTo(TestPojo("Test", date)) } @Autowired @@ -119,7 +118,7 @@ class KotlinDslTests { @Test fun `fixed subscriber channel`() { assertThat(MessagingTemplate().convertSendAndReceive(this.fixedSubscriberInput, "test", String::class.java)) - .isEqualTo("test") + .isEqualTo("test") } @Autowired @@ -151,19 +150,19 @@ class KotlinDslTests { val fluxChannel = FluxMessageChannel() val verifyLater = - StepVerifier - .create(Flux.from(fluxChannel).map { it.payload }.cast(Integer::class.java)) - .expectNext(Integer(4), Integer(6)) - .thenCancel() - .verifyLater() + StepVerifier + .create(Flux.from(fluxChannel).map { it.payload }.map { it.toString().toInt() }) + .expectNext(4, 6) + .thenCancel() + .verifyLater() val publisher = Flux.just(2, 3).map { GenericMessage(it) } val integrationFlow = - integrationFlow(publisher) { - transform>({ it.payload * 2 }) { id("foo") } - channel(fluxChannel) - } + integrationFlow(publisher) { + transform>({ it.payload * 2 }) { id("foo") } + channel(fluxChannel) + } val registration = this.integrationFlowContext.registration(integrationFlow).register() @@ -198,9 +197,9 @@ class KotlinDslTests { fun `Scatter-Gather`() { val replyChannel = QueueChannel() val request = - MessageBuilder.withPayload("foo") - .setReplyChannel(replyChannel) - .build() + MessageBuilder.withPayload("foo") + .setReplyChannel(replyChannel) + .build() this.scatterGatherFlowInput.send(request) val bestQuoteMessage = replyChannel.receive(10000) assertThat(bestQuoteMessage).isNotNull() @@ -212,9 +211,9 @@ class KotlinDslTests { fun `no reply from handle`() { val payloadReference = AtomicReference() val integrationFlow = - integrationFlow("handlerInputChanenl") { - handle { payload, _ -> payloadReference.set(payload) } - } + integrationFlow("handlerInputChanenl") { + handle { payload, _ -> payloadReference.set(payload) } + } val registration = this.integrationFlowContext.registration(integrationFlow).register() @@ -231,89 +230,89 @@ class KotlinDslTests { @Bean(PollerMetadata.DEFAULT_POLLER) fun defaultPoller() = - Pollers.fixedDelay(100).maxMessagesPerPoll(1).get() + Pollers.fixedDelay(100).maxMessagesPerPoll(1).get() @Bean fun convertFlow() = - integrationFlow("convertFlowInput") { - convert() - convert { id("kotlinConverter") } - handle { m -> (m.headers[MessageHeaders.REPLY_CHANNEL] as MessageChannel).send(m) } - } + integrationFlow("convertFlowInput") { + convert() + convert { id("kotlinConverter") } + handle { m -> (m.headers[MessageHeaders.REPLY_CHANNEL] as MessageChannel).send(m) } + } @Bean fun functionFlow() = - integrationFlow>({ beanName("functionGateway") }) { - transform(Transformers.objectToString()) { id("objectToStringTransformer") } - transform { it.toUpperCase() } - split> { it.payload } - split({ it }) { id("splitterEndpoint") } - resequence() - aggregate { - id("aggregator") - outputProcessor { it.one } - } + integrationFlow>({ beanName("functionGateway") }) { + transform(Transformers.objectToString()) { id("objectToStringTransformer") } + transform { it.toUpperCase() } + split> { it.payload } + split({ it }) { id("splitterEndpoint") } + resequence() + aggregate { + id("aggregator") + outputProcessor { it.one } } + } @Bean fun functionFlow2() = - integrationFlow> { - transform { it.toLowerCase() } - filter(UnexpiredMessageSelector()) - route, Any?>({ null }) { defaultOutputToParentFlow() } - route> { m -> m.headers.replyChannel } - } + integrationFlow> { + transform { it.toLowerCase() } + filter(UnexpiredMessageSelector()) + route, Any?>({ null }) { defaultOutputToParentFlow() } + route> { m -> m.headers.replyChannel } + } @Bean fun messageSourceFlow() = - integrationFlow(MessageProcessorMessageSource { "testSource" }, - { poller { it.trigger(OnlyOnceTrigger()) } }) { - publishSubscribe(PublishSubscribeChannel(), - { - channel { queue("fromSupplierQueue") } - }, - { - log(LoggingHandler.Level.WARN) { "From second subscriber: ${it.payload}"} - }) - } + integrationFlow(MessageProcessorMessageSource { "testSource" }, + { poller { it.trigger(OnlyOnceTrigger()) } }) { + publishSubscribe(PublishSubscribeChannel(), + { + channel { queue("fromSupplierQueue") } + }, + { + log(LoggingHandler.Level.WARN) { "From second subscriber: ${it.payload}" } + }) + } @Bean fun messageSourceFlow2() = - integrationFlow(MessageProcessorMessageSource { "testSource2" }) { - channel { queue("fromSupplierQueue2") } - } + integrationFlow(MessageProcessorMessageSource { "testSource2" }) { + channel { queue("fromSupplierQueue2") } + } @Bean fun fixedSubscriberFlow() = - integrationFlow("fixedSubscriberInput", true) { - log(LoggingHandler.Level.WARN) { it.payload } - transform("payload") { id("spelTransformer") } - } + integrationFlow("fixedSubscriberInput", true) { + log(LoggingHandler.Level.WARN) { it.payload } + transform("payload") { id("spelTransformer") } + } @Bean fun flowFromSupplier() = - integrationFlow({ "testSupplier" }) { - channel { queue("testSupplierResult") } - } + integrationFlow({ "testSupplier" }) { + channel { queue("testSupplierResult") } + } @Bean fun flowFromSupplier2() = - integrationFlow({ "testSupplier2" }, - { poller { it.trigger(OnlyOnceTrigger()) } }) { - filter> { m -> m.payload is String } - channel { queue("testSupplierResult2") } - } + integrationFlow({ "testSupplier2" }, + { poller { it.trigger(OnlyOnceTrigger()) } }) { + filter> { m -> m.payload is String } + channel { queue("testSupplierResult2") } + } @Bean fun flowLambda() = - integrationFlow { - filter({ it === "test" }) { id("filterEndpoint") } - wireTap { - channel { queue("wireTapChannel") } - } - delay("delayGroup") { defaultDelay(100) } - transform { it.toUpperCase() } + integrationFlow { + filter({ it === "test" }) { id("filterEndpoint") } + wireTap { + channel { queue("wireTapChannel") } } + delay("delayGroup") { defaultDelay(100) } + transform { it.toUpperCase() } + } /* @@ -337,23 +336,25 @@ class KotlinDslTests { }*/ @Bean fun scatterGatherFlow() = - integrationFlow { - scatterGather( - { - applySequence(true) - recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } - recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } - recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } - }, - { - releaseStrategy { - it.size() == 3 || it.messages.stream().anyMatch { it.payload as Double > 5 } - } - }) + integrationFlow { + scatterGather( { - gatherTimeout(10_000) - } + applySequence(true) + recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } + recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } + recipientFlow({ true }) { handle { _, _ -> Math.random() * 10 } } + }, + { + releaseStrategy { + it.size() == 3 || it.messages.stream().anyMatch { it.payload as Double > 5 } + } + }) + { + gatherTimeout(10_000) } + } + + } data class TestPojo(val name: String?, val date: Date?) diff --git a/src/reference/asciidoc/delayer.adoc b/src/reference/asciidoc/delayer.adoc index c707bd038d..5ec81c2232 100644 --- a/src/reference/asciidoc/delayer.adoc +++ b/src/reference/asciidoc/delayer.adoc @@ -25,39 +25,10 @@ The following example delays all messages by three seconds: If you need to determine the delay for each message, you can also provide the SpEL expression by using the 'expression' attribute, as the following expression shows: -==== -[source,xml] ----- - ----- -==== - -In the preceding example, the three-second delay applies only when the expression evaluates to null for a given inbound message. -If you want to apply a delay only to messages that have a valid result of the expression evaluation, you can use a 'default-delay' of `0` (the default). -For any message that has a delay of `0` (or less), the message is sent immediately, on the calling thread. - -The following example shows the Java configuration equivalent of the preceding example: ==== -[source, java] ----- -@ServiceActivator(inputChannel = "input") -@Bean -public DelayHandler delayer() { - DelayHandler handler = new DelayHandler("delayer.messageGroupId"); - handler.setDefaultDelay(3_000L); - handler.setDelayExpressionString("headers['delay']"); - handler.setOutputChannelName("output"); - return handler; -} ----- -==== - -The following example shows the Java DSL equivalent of the preceding example: - -==== -[source, java] +[source, java, role="primary"] +.Java DSL ---- @Bean public IntegrationFlow flow() { @@ -69,8 +40,44 @@ public IntegrationFlow flow() { .get(); } ---- +[source, kotlin, role="secondary"] +.Kotlin DSL +---- +@Bean +fun flow() = + integrationFlow("input") { + delay("delayer.messageGroupId") { + defaultDelay(3000L) + delayExpression("headers['delay']") + } + channel("output") + } +---- +[source, java, role="secondary"] +.Java +---- +@ServiceActivator(inputChannel = "input") +@Bean +public DelayHandler delayer() { + DelayHandler handler = new DelayHandler("delayer.messageGroupId"); + handler.setDefaultDelay(3_000L); + handler.setDelayExpressionString("headers['delay']"); + handler.setOutputChannelName("output"); + return handler; +} +---- +[source, xml, role="secondary"] +.XML +---- + +---- ==== +In the preceding example, the three-second delay applies only when the expression evaluates to null for a given inbound message. +If you want to apply a delay only to messages that have a valid result of the expression evaluation, you can use a 'default-delay' of `0` (the default). +For any message that has a delay of `0` (or less), the message is sent immediately, on the calling thread. + NOTE: The XML parser uses a message group ID of `.messageGroupId`. TIP: The delay handler supports expression evaluation results that represent an interval in milliseconds (any `Object` whose `toString()` method produces a value that can be parsed into a `Long`) as well as `java.util.Date` instances representing an absolute time. @@ -182,12 +189,16 @@ These operations can be invoked through a `Control Bus` command, as the followin ---- Message delayerReschedulingMessage = MessageBuilder.withPayload("@'delayer.handler'.reschedulePersistedMessages()").build(); - controlBusChannel.send(delayerReschedulingMessage); +controlBusChannel.send(delayerReschedulingMessage); ---- ==== NOTE: For more information regarding the message store, JMX, and the control bus, see <<./system-management.adoc#system-management-chapter,System Management>>. +Starting with version 5.3.7, if a transaction is active when a message is stored into a `MessageStore`, the release task is scheduled in a `TransactionSynchronization.afterCommit()` callback. +This is necessary to prevent a race condition, where the scheduled release could run before the transaction has committed, and the message is not found. +In this case, the message will be released after the delay, or after the transaction commits, whichever is later. + [[delayer-release-failures]] ==== Release Failures