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 2d443f9143
commit c5e24fb301
4 changed files with 189 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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