Spring Transactions are not rolled back when a publisher is returned with error in a suspend function

Affects: v5.3.7

---

I'm using Kotlin Coroutines to build a Spring Webflux application and recently I got an issue working with suspending functions and Project Reactor.

When I return a publisher with an error in a suspend function the transaction is not rolled back.

e.g.

@Transactional
suspend fun save(name: String): Flux<Employee> {
  repository.save(Employee(name = name)).awaitSingle()
  return Flux.error(RuntimeException())
}

But when I change the return type to Flow the transaction is rolled back.

e.g.

@Transactional
suspend fun save(name: String): Flow<Employee> {
  repository.save(Employee(name = name)).awaitSingle()
  return Flux.error<Employee>(RuntimeException()).asFlow()
}

Idk if I'm missing something here so please let me know.

I've created a small project with some test cases to reproduce this issue here.

Comment From: poutsma

The cause for the difference between the way Flows are treated is this line.

@sdeleuze I have a proposed fix for this issue here. Could you please review?

Comment From: sdeleuze

If I understand the proposed changes right, this would add special handling of Mono and other Publisher (including Flux) in addition to Flow for suspending functions.

I think this is a change that generally makes sense, not only fixing this use case on Reactive transaction, but also improving the DevXP for people returning Mono and Flux in suspending functions (not super common, but we had that use case reported in the past).

That said, if I am not mistaken, this would be a breaking change where we would need feedback, so I recommend shipping that as part of the next Spring Framework 6 milestone.