How to add transaction support to Java DSL integration flows

Question

I have to add a transaction support to an integration flow. Let's assume that there are 3 transformers. The first and third transformers should be done within the same transaction, but the second one shouldn't be done within any transaction. Thus, if an error occurs in the third transformer, all changes from the first and third transformers should not be committed but changes from the second transformer should be committed. How can I do that? I tried to add .transform(FirstMessageTransformer, e -> e.transactional(true))but then all transformers are done within a transaction. I tried also to add .transform(FirstMessageTransformer, e -> e.transactional(false))but it doesn't seem to work well, because the changes are committed for all tranformers, even if an exception occurs.

@Bean
public IntegrationFlow myMessageFromMessageAmqpInboundFlow() {
    return IntegrationFlows.from(myInboundChannel)
            .transform(FirstMessageTransformer)
            .transform(SecondMessageTransformer)
            .transform(ThirdMessageTransformer)
            .channel(anOutputChannel)
            .get();
}

Answer 1

Try like this:

        .transform(FirstMessageTransformer, e -> e.transactional(true))
        .transform(SecondMessageTransformer, 
                        e -> e.transactional(
                                     new TransactionInterceptorBuilder()
                                             .transactionManager(txManager)
                                             .propagation(Propagation.NOT_SUPPORTED)
                                             .build()))
        .transform(ThirdMessageTransformer)

This way you will have a transaction for the whole sub-flow starting with FirstMessageTransformer and that Propagation.NOT_SUPPORTED will say for the SecondMessageTransformer to suspend the current transaction and perform only this MessageHandler out of transaction. After finishing work with the SecondMessageTransformer, the original transaction should resume and continue for the rest of the flow.