title: How do I use a Transaction in a Reactive Flow in Spring Integration?
body: <p>I am querying a database for an item using R2DBC and Spring Integration.  I want to extend the transaction boundary a bit to include a handler - if the handler fails I want to roll back the database operation.  But I'm having difficulty even establishing transactionality explicitly in my integration flow.  The flow is defined as</p>
  <pre class="lang-java prettyprint-override"><code>@Bean
  IntegrationFlow flow(R2dbcEntityTemplate r2dbcEntityTemplate) {
      return IntegrationFlows
          .from(R2dbc.inboundChannelAdapter(r2dbcEntityTemplate,
                  &quot;SELECT * FROM events LIMIT 1&quot;)
              .expectSingleResult(true)
              .payloadType(Event.class),
              e - &gt; e.poller(Pollers.cron(&quot;30/2 * * * * *&quot;)
                  .transactional(transactionManager)))
          .channel(MessageChannels.flux())
          .handle(Mono.class, (payload, headers) - &gt; doSomethingUsingSameTransaction(payload), e - &gt; e.async(true))
          .channel(MessageChannels.queue(&quot;queue&quot;))
          .get();
  }
  </code></pre>
  <p>where the transaction manager is obtained in this way:</p>
  <pre class="lang-java prettyprint-override"><code>@Bean
  ReactiveTransactionManager transactionManager(ConnectionFactory connectionFactory) {
      return new R2dbcTransactionManager(connectionFactory);
  }
  </code></pre>
  <p>When trying this, I get an exception:</p>
  <pre><code>reactor.core.Exceptions$ErrorCallbackNotImplemented: 
  org.springframework.messaging.MessagingException: 
      nested exception is java.lang.IllegalStateException: 
      Cannot apply reactive transaction to non-reactive return type: class java.lang.Object
  </code></pre>
  <p>Looking at the Spring Integration code reveals the issue.</p>
  <p>A TransactionInterceptor is added as advice to the polling task created in <code>AbstractPollingEndpoint#createPollingTask</code>.  That task is a <code>Callable&lt;Message&lt;?&gt;&gt;</code>.</p>
  <p>Soon, the poller (a reactive implementation) triggers and <code>AbstractPollingEndpoint#pollForMessage</code> is invoked - causing the advice <code>TransactionInterceptor#invoke</code> to be called. It calls <code>TransactionAspectSupport#invokeWithinTransaction</code>, which determines that the transaction manager is a reactive one, and that the polling task does not return a reactive type (it's the Object returned from the Callable) - and throws the exception.</p>
  <p>The failure occurs before handing off to the channel or anything after it.</p>
  <p>It makes sense, since the R2DBC endpoint returns a <code>Message&lt;Flux&lt;Event&gt;&gt;</code>.</p>
  <p>So, I'm left wondering how to access the transaction semantics for a reactive flow -</p>
  <p>In other words, I'm expecting that the transaction boundaries would start with the poll and end on its return - and that the transaction applies for the lifetime of the reactive context (subscription).  The actual application does leverage FluxChannels and the handler is a consumer of one of those channels.</p>
  <p>I realize that I want to keep those boundaries slim, but in the use case I have, I need to perform a locking operation on a table row, perform an external task and release the lock within the scope of a transaction.</p>
  <p>The options I have seem to be :</p>
  <ol>
  <li>Start the Transaction at the start of the flow - this likely suffers from the same problem I am having now, since there's no indication this is a reactive flow:</li>
  </ol>
  <pre class="lang-java prettyprint-override"><code>    @Transactional
      @Bean
      IntegrationFlow flow(...
  </code></pre>
  <ol start="2">
  <li><p>Try something like
  <code>.transactional(new TransactionInterceptorBuilder().build())</code></p>
  <p>which will create a transaction if needed - but I'm not sure it will be aware of the reactive transaction manager and its state...</p>
  </li>
  <li><p>Use a compensating action - recalling the former state of the data in the row</p>
  </li>
  <li><p>Use my own handler to encapsulate the database operation and other actions to wrap all of the operations together.</p>
  </li>
  <li><p>Manage the transaction myself</p>
  </li>
  <li><p>Configure the flow differently based on help from this forum.</p>
  </li>
  </ol>
  <p>Is there a way to extend the transactional boundaries in defining the flow?  Or what would be an effective way to accomplish the task?</p>
  <p>Thank you very much in advance.</p>
  <p>This complete test illustrates the problem:</p>
  <pre class="lang-java prettyprint-override"><code>@Slf4j
  @SpringJUnitConfig
  @DirtiesContext
  public class R2dbcTransactionalTest {
  
      @Autowired
      DatabaseClient client;
  
      R2dbcEntityTemplate entityTemplate;
  
      @Qualifier(&quot;queue&quot;)
      @Autowired
      QueueChannel validationChannel;
  
      @BeforeEach
      public void setup() {
          this.entityTemplate = new R2dbcEntityTemplate(this.client, H2Dialect.INSTANCE);
  
          List &lt; String &gt; statements =
              Arrays.asList(
                  &quot;DROP TABLE IF EXISTS events;&quot;,
                  &quot;CREATE TABLE events (id INT AUTO_INCREMENT NOT NULL, details VARCHAR2 NOT NULL, timestamp TIMESTAMP NOT NULL);&quot;);
  
          statements.forEach(it - &gt; this.client.sql(it)
              .fetch()
              .rowsUpdated()
              .as(StepVerifier::create)
              .expectNextCount(1)
              .verifyComplete());
      }
  
      @Test
      public void validateSuccessfulIntegrationFlow() {
          this.entityTemplate.insert(new Event(Instant.now(), &quot;Event details&quot;))
              .then()
              .as(StepVerifier::create)
              .expectComplete()
              .verify(Duration.ofSeconds(1));
  
          // Validate string
          final Message &lt;&lt; ? &gt; message = validationChannel.receive();
          assertThat(message.getPayload()).isEqualTo(&quot;Event details&quot;);
          assertThat(message.getHeaders()).containsKey(&quot;foo&quot;);
      }
  
      @Import(R2dbcDatabaseConfiguration.class)
      @Configuration
      @EnableIntegration
      static class SpringIntegrationConfiguration {
  
          @Autowired
          R2dbcEntityTemplate r2dbcEntityTemplate;
  
          @Autowired
          ReactiveTransactionManager transactionManager;
  
          @Bean
          IntegrationFlow flow(R2dbcEntityTemplate r2dbcEntityTemplate) {
              return IntegrationFlows
                  .from(R2dbc.inboundChannelAdapter(r2dbcEntityTemplate,
                          &quot;SELECT * FROM events LIMIT 1&quot;)
                      .expectSingleResult(true)
                      .payloadType(Event.class),
                      e - &gt; e.poller(Pollers.cron(&quot;30/2 * * * * *&quot;)
                          .transactional(transactionManager)))
                  .channel(MessageChannels.flux())
                  .handle(Mono.class, (payload, headers) - &gt; doSomethingUsingSameTransaction(payload), e - &gt; e.async(true))
                  .channel(MessageChannels.queue(&quot;queue&quot;))
                  .get();
          }
  
          public Mono &lt; Message &lt; String &gt;&gt; doSomethingUsingSameTransaction(Mono &lt; Event &gt; eventMono) {
              return eventMono.map(event - &gt;
                  MessageBuilder
                  .withPayload(event.getDetails())
                  .setHeader(&quot;foo&quot;, &quot;baz&quot;)
                  .build());
          }
      }
  
      @Configuration
      @EnableR2dbcRepositories(basePackages = &quot;org.springframework.integration.r2dbc.repository&quot;)
      static class R2dbcDatabaseConfiguration extends AbstractR2dbcConfiguration {
  
          @Bean
          @Override
          public ConnectionFactory connectionFactory() {
              return createConnectionFactory();
          }
  
          public ConnectionFactory createConnectionFactory() {
              return new H2ConnectionFactory(H2ConnectionConfiguration.builder()
                  .inMemory(&quot;r2dbc&quot;)
                  .username(&quot;sa&quot;)
                  .password(&quot;&quot;)
                  .option(&quot;DB_CLOSE_DELAY=-1&quot;).build());
          }
  
          @Bean
          public DatabaseClient databaseClient(ConnectionFactory connectionFactory) {
              return DatabaseClient.create(connectionFactory);
          }
  
          @Bean
          public R2dbcEntityTemplate r2dbcEntityTemplate(DatabaseClient databaseClient) {
              return new R2dbcEntityTemplate(databaseClient, H2Dialect.INSTANCE);
          }
  
          @Bean
          ReactiveTransactionManager transactionManager(ConnectionFactory connectionFactory) {
              return new R2dbcTransactionManager(connectionFactory);
          }
  
      }
  
      @Table(&quot;events&quot;)
      @Getter
      @Setter
      @RequiredArgsConstructor
      static class Event {
          @Id
          private Integer id;
  
          @NonNull
          public Instant timestamp;
          @NonNull
          public String details;
  
      }
  
  }
  </code></pre>