title: Spring cloud stream : StreamBridge and Transaction
body: <p>I would like to have a spring cloud stream listener handling a fulle transaction regarding all messages sent in it. Or, all message sent manually with StreamBridge in the function are commited even if there is an exception after.</p>
  <p>This is my lib versions :</p>
  <pre><code>spring : 2.5.5
  spring cloud stream : 3.1.4
  spring cloud stream rabbit binder : 3.1.4
  </code></pre>
  <p>My spring cloud stream conf :</p>
  <pre class="lang-yaml prettyprint-override"><code>spring:
    cloud:
      function:
        definition: test
      stream:
        rabbit:
          default:
            producer:
              transacted: true
            consumer:
              transacted: true
          bindings:
            test-in-0:
              consumer:
                queueNameGroupOnly: true
                receive-timeout: 500
                transacted: true
            test-out-0:
              producer:
                queueNameGroupOnly: true
                transacted: true
            other-out-0:
              producer:
                queueNameGroupOnly: true
                transacted: true
        bindings:
          test-in-0:
            destination: test.request
            group: test.request
            consumer:
              requiredGroups: test.request
              maxAttempts: 1
          test-out-0:
            destination: test.response
            group:  test.response
            producer:
              requiredGroups:  test.response
          other-out-0:
            destination: test.other.request
            group: test.other.request
            producer:
              requiredGroups: test.other.request
  </code></pre>
  <p>My test java code :</p>
  <pre class="lang-java prettyprint-override"><code>@Configuration
  public class TestSender {
      @Bean
      public Function&lt;Message&lt;TestRequest&gt;, Message&lt;String&gt;&gt; test(Service service) {
          return (request) -&gt; service.run(request.getPayload().getContent());
      }
  }
  </code></pre>
  <pre class="lang-java prettyprint-override"><code>@Component
  @Transactional
  public class Service {
      private static final Logger LOGGER = LoggerFactory.getLogger(Service.class);
      StreamBridge bridge;
      IWorker worker;
      public Service(StreamBridge bridge, IWorker worker) {
          this.bridge = bridge;
          this.worker = worker;
      }
  
      @Transactional
      public Message&lt;String&gt; run(String message) {
          LOGGER.info(&quot;Processing {}&quot;, message);
          bridge.send(&quot;other-out-0&quot;, MessageBuilder.withPayload(&quot;test&quot;)
                  .setHeader(&quot;toto&quot;, &quot;titi&quot;).build());
          if (message.equals(&quot;error&quot;)) {
              throw new RuntimeException(&quot;test error&quot;);
          }
          return MessageBuilder.withPayload(&quot;test&quot;)
                  .setHeader(&quot;toto&quot;, &quot;titi&quot;).build();
      }
  }
  </code></pre>
  <p>Test class to run :</p>
  <pre class="lang-java prettyprint-override"><code>@SpringBootApplication
  public class EmptyWorkerApplication {
  
      private static final Logger LOGGER = LoggerFactory.getLogger(EmptyWorkerApplication.class);
  
      public static void main(String[] args) {
          SpringApplication.run(EmptyWorkerApplication.class, args);
      }
  
      @Bean
      public ApplicationRunner runner(RabbitTemplate template) {
          return args -&gt; {
              LOGGER.info(&quot;Sending messages ...&quot;);
  template.convertAndSend(&quot;test.request&quot;, &quot;#&quot;,
                                  org.springframework.amqp.core.MessageBuilder.withBody(
                                                  &quot;{\&quot;content\&quot;:\&quot;toto\&quot;}&quot;.getBytes(StandardCharsets.UTF_8))
                                          .setContentType(&quot;application/json&quot;)
                                          .build());
          template.convertAndSend(&quot;test.request&quot;, &quot;#&quot;,
                                  org.springframework.amqp.core.MessageBuilder.withBody(
                                          &quot;{\&quot;content\&quot;:\&quot;error\&quot;}&quot;.getBytes(StandardCharsets.UTF_8))
                                          .setContentType(&quot;application/json&quot;)
                                          .build());
          template.convertAndSend(&quot;test.request&quot;, &quot;#&quot;,
                                  org.springframework.amqp.core.MessageBuilder.withBody(
                                          &quot;{\&quot;content\&quot;:\&quot;titi\&quot;}&quot;.getBytes(StandardCharsets.UTF_8))
                                          .setContentType(&quot;application/json&quot;)
                                          .build());
          };
      }
  </code></pre>
  <p>I also added a TransactionManager :</p>
  <pre class="lang-java prettyprint-override"><code>@Configuration
  @EnableTransactionManagement
  public class TransactionManagerConfiguration {
  
      @Bean(name = &quot;transactionManager&quot;)
      public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory cf) {
          RabbitTransactionManager manager = new RabbitTransactionManager(cf);
          return manager;
      }
  
  }
  </code></pre>
  <p>With this example at the end i have in my rabbit queues :</p>
  <p><a href="https://i.stack.imgur.com/1fenl.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/1fenl.png" alt="result" /></a></p>
  <p>Or i was expected to have only 2 message on test.other.request. What i'm doing wrong ?</p>
  <p><strong>EDIT 1</strong></p>
  <p>Tryed code :</p>
  <pre class="lang-java prettyprint-override"><code>@Component(&quot;myfunction&quot;)
  public class Myfunction implements Consumer&lt;String&gt; {
  
      private final StreamBridge streamBridge;
  
      public Myfunction(StreamBridge streamBridge) {
          this.streamBridge = streamBridge;
      }
  
      @Override
      @Transactional
      public void accept(String request) {
          this.streamBridge.send(&quot;myfunction-out-0&quot;, request);
          if (request.equals(&quot;error&quot;)) {
              throw new RuntimeException(&quot;test error&quot;);
          }
      }
  }
  </code></pre>
  <pre class="lang-java prettyprint-override"><code>@SpringBootApplication
  public class EmptyWorkerApplication {
  
      public static void main(String[] args) {
          SpringApplication.run(EmptyWorkerApplication.class, args);
      }
  
      @Bean
      public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory cf) {
          RabbitTransactionManager manager = new RabbitTransactionManager(cf);
          return manager;
      }
  
      @Bean
      public ApplicationRunner runner(RabbitTemplate template) {
          return args -&gt; {
  template.convertAndSend(&quot;test.request&quot;, &quot;#&quot;,
                                  org.springframework.amqp.core.MessageBuilder.withBody(
                                                  &quot;test&quot;.getBytes(StandardCharsets.UTF_8))
                                          .setContentType(&quot;text/plain&quot;)
                                          .build());
          template.convertAndSend(&quot;test.request&quot;, &quot;#&quot;,
                                  org.springframework.amqp.core.MessageBuilder.withBody(
                                          &quot;error&quot;.getBytes(StandardCharsets.UTF_8))
                                          .setContentType(&quot;text/plain&quot;)
                                          .build());
          template.convertAndSend(&quot;test.request&quot;, &quot;#&quot;,
                                  org.springframework.amqp.core.MessageBuilder.withBody(
                                          &quot;test&quot;.getBytes(StandardCharsets.UTF_8))
                                          .setContentType(&quot;text/plain&quot;)
                                          .build());
          };
      }
  }
  </code></pre>
  <pre class="lang-yaml prettyprint-override"><code>spring:
    rabbitmq:
      host: xx
      port: xx
      username: xx
      password: xx
      virtual-host: xx
    cloud:
      function:
        definition: myfunction
      stream:
        rabbit:
           bindings:
             myfunction-in-0:
               queueNameGroupOnly: true
             myfunction-out-0:
               queueNameGroupOnly: true
               transacted: true
        bindings:
          myfunction-in-0:
            destination: test.request
            group: test.request
            consumer:
              requiredGroups: test.request
              autoBindDlq: true
              maxAttempts: 1
          myfunction-out-0:
            destination: test.response
            group:  test.response
            producer:
              requiredGroups:  test.response
  </code></pre>
  <p><strong>EDIT 2 :</strong></p>
  <p>I finally succeeded to make it work.
  My mistake was to set the property
  <code>spring.cloud.stream.rabbit.bindings.myfunction-in-0.consumer.transacted=true</code>
  instead of
  <code>spring.cloud.stream.rabbit.bindings.myfunction-in.consumer.transacted=true</code></p>
  <p>I actually dont understand the difference and i didn't found any explanation in both spring cloud stream and spring cloud rabbit binder documentation.</p>