Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3521: Delayer: schedule release task with TX (#3525)

Browse Source 
* 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 <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2021-03-24 11:12:03 -04:00
committed by Gary Russell
parent d016dd5b6f
commit a7843af045
4 changed files with 189 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -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) {

57
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java
View File

@@ -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<String>("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<String>("1"));
inputB.send(new GenericMessage<String>("2"));
inputB.send(new GenericMessage<String>("3"));
inputB.send(new GenericMessage<String>("4"));
inputB.send(new GenericMessage<String>("5"));
inputB.send(new GenericMessage<String>("6"));
inputB.send(new GenericMessage<String>("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<String>("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<String>("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");
}

189
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
View File

@@ -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<Message<Int>>({ it.payload * 2 }) { id("foo") }
channel(fluxChannel)
}
integrationFlow(publisher) {
transform<Message<Int>>({ 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<String>()
val integrationFlow =
integrationFlow("handlerInputChanenl") {
handle<String> { payload, _ -> payloadReference.set(payload) }
}
integrationFlow("handlerInputChanenl") {
handle<String> { 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<TestPojo>()
convert<TestPojo> { id("kotlinConverter") }
handle { m -> (m.headers[MessageHeaders.REPLY_CHANNEL] as MessageChannel).send(m) }
}
integrationFlow("convertFlowInput") {
convert<TestPojo>()
convert<TestPojo> { id("kotlinConverter") }
handle { m -> (m.headers[MessageHeaders.REPLY_CHANNEL] as MessageChannel).send(m) }
}
@Bean
fun functionFlow() =
integrationFlow<Function<ByteArray, String>>({ beanName("functionGateway") }) {
transform(Transformers.objectToString()) { id("objectToStringTransformer") }
transform<String> { it.toUpperCase() }
split<Message<*>> { it.payload }
split<String>({ it }) { id("splitterEndpoint") }
resequence()
aggregate {
id("aggregator")
outputProcessor { it.one }
}
integrationFlow<Function<ByteArray, String>>({ beanName("functionGateway") }) {
transform(Transformers.objectToString()) { id("objectToStringTransformer") }
transform<String> { it.toUpperCase() }
split<Message<*>> { it.payload }
split<String>({ it }) { id("splitterEndpoint") }
resequence()
aggregate {
id("aggregator")
outputProcessor { it.one }
}
}
@Bean
fun functionFlow2() =
integrationFlow<Function<*, *>> {
transform<String> { it.toLowerCase() }
filter(UnexpiredMessageSelector())
route<Message<*>, Any?>({ null }) { defaultOutputToParentFlow() }
route<Message<*>> { m -> m.headers.replyChannel }
}
integrationFlow<Function<*, *>> {
transform<String> { it.toLowerCase() }
filter(UnexpiredMessageSelector())
route<Message<*>, Any?>({ null }) { defaultOutputToParentFlow() }
route<Message<*>> { m -> m.headers.replyChannel }
}
@Bean
fun messageSourceFlow() =
integrationFlow(MessageProcessorMessageSource { "testSource" },
{ poller { it.trigger(OnlyOnceTrigger()) } }) {
publishSubscribe(PublishSubscribeChannel(),
{
channel { queue("fromSupplierQueue") }
},
{
log<Any>(LoggingHandler.Level.WARN) { "From second subscriber: ${it.payload}"}
})
}
integrationFlow(MessageProcessorMessageSource { "testSource" },
{ poller { it.trigger(OnlyOnceTrigger()) } }) {
publishSubscribe(PublishSubscribeChannel(),
{
channel { queue("fromSupplierQueue") }
},
{
log<Any>(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<Any>(LoggingHandler.Level.WARN) { it.payload }
transform("payload") { id("spelTransformer") }
}
integrationFlow("fixedSubscriberInput", true) {
log<Any>(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<Message<*>> { m -> m.payload is String }
channel { queue("testSupplierResult2") }
}
integrationFlow({ "testSupplier2" },
{ poller { it.trigger(OnlyOnceTrigger()) } }) {
filter<Message<*>> { m -> m.payload is String }
channel { queue("testSupplierResult2") }
}
@Bean
fun flowLambda() =
integrationFlow {
filter<String>({ it === "test" }) { id("filterEndpoint") }
wireTap {
channel { queue("wireTapChannel") }
}
delay("delayGroup") { defaultDelay(100) }
transform<String> { it.toUpperCase() }
integrationFlow {
filter<String>({ it === "test" }) { id("filterEndpoint") }
wireTap {
channel { queue("wireTapChannel") }
}
delay("delayGroup") { defaultDelay(100) }
transform<String> { it.toUpperCase() }
}
/*
@@ -337,23 +336,25 @@ class KotlinDslTests {
}*/
@Bean
fun scatterGatherFlow() =
integrationFlow {
scatterGather(
{
applySequence(true)
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
},
{
releaseStrategy {
it.size() == 3 || it.messages.stream().anyMatch { it.payload as Double > 5 }
}
})
integrationFlow {
scatterGather(
{
gatherTimeout(10_000)
}
applySequence(true)
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> Math.random() * 10 } }
recipientFlow<Any>({ true }) { handle<Any> { _, _ -> 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?)

75
src/reference/asciidoc/delayer.adoc
View File

@@ -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]
----
<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" expression="headers['delay']"/>
----
====
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
----
<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" expression="headers['delay']"/>
----
====
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 `<beanName>.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<String> 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