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 GitHub
parent 7e9552974c
commit 67c5cf7458
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

@@ -52,6 +52,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;
@@ -441,7 +443,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) {