R2dbc transactions not working properly when using coroutines

Steps to Reproduce

When using the @Transactional on a method that returns a kotlin flow i am getting a NoTransactionException

@Transactional(Transactional.TxType.MANDATORY)
@R2dbcRepository(dialect = Dialect.POSTGRES)
interface RecordTransactionalCoroutineRepository : CoroutineCrudRepository<Record, UUID>

@Transactional
open fun saveAllUsingCoroutines(records: Iterable<Record>): Flow<Record> = coroutineRepository.saveAll(records)

The same error also happens when using ReactiveStreamsCrudRepository or when using declarative transactions

Expected Behaviour

The transaction should be propagated on the context and it should be committed only after the last item on the flow is collected

Actual Behaviour

Fails with the given stack

14:37:54.217 [reactor-tcp-epoll-2] WARN  i.m.d.r.o.DefaultR2dbcRepositoryOperations - Rolling back transaction: RecordTransactionalService.saveAllUsingCoroutines on error: Expected an existing transaction, but none was found in the Reactive context. for dataSource default
io.micronaut.transaction.exceptions.NoTransactionException: Expected an existing transaction, but none was found in the Reactive context.
	at io.micronaut.data.r2dbc.operations.DefaultR2dbcRepositoryOperations.lambda$withTransaction$15(DefaultR2dbcRepositoryOperations.java:410)
	at reactor.core.publisher.FluxDeferContextual.subscribe(FluxDeferContextual.java:49)
	at reactor.core.publisher.Flux.subscribe(Flux.java:8660)
	at kotlinx.coroutines.reactive.PublisherAsFlow.collectImpl(ReactiveFlow.kt:94)
	at kotlinx.coroutines.reactive.PublisherAsFlow.collect(ReactiveFlow.kt:79)
	at kotlinx.coroutines.reactive.FlowSubscription.consumeFlow(ReactiveFlow.kt:275)
	at kotlinx.coroutines.reactive.FlowSubscription.flowProcessing(ReactiveFlow.kt:209)
	at kotlinx.coroutines.reactive.FlowSubscription.access$flowProcessing(ReactiveFlow.kt:187)
	at kotlinx.coroutines.reactive.FlowSubscription$createInitialContinuation$1$1.invoke(ReactiveFlow.kt:204)
	at kotlinx.coroutines.reactive.FlowSubscription$createInitialContinuation$1$1.invoke(ReactiveFlow.kt:204)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt$createCoroutineUnintercepted$$inlined$createCoroutineFromSuspendFunction$IntrinsicsKt__IntrinsicsJvmKt$2.invokeSuspend(IntrinsicsJvm.kt:205)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.internal.DispatchedContinuationKt.resumeCancellableWith(DispatchedContinuation.kt:367)
	at kotlinx.coroutines.internal.DispatchedContinuationKt.resumeCancellableWith$default(DispatchedContinuation.kt:278)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:18)
	at kotlinx.coroutines.reactive.FlowSubscription$createInitialContinuation$$inlined$Continuation$1.resumeWith(Continuation.kt:162)
	at kotlinx.coroutines.reactive.FlowSubscription.request(ReactiveFlow.kt:267)
	at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.request(FluxContextWrite.java:136)
	at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.request(FluxUsingWhen.java:319)
	at reactor.core.publisher.Operators$DeferredSubscription.set(Operators.java:1717)
	at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.onSubscribe(FluxUsingWhen.java:409)
	at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onSubscribe(FluxContextWrite.java:101)
	at kotlinx.coroutines.reactive.FlowAsPublisher.subscribe(ReactiveFlow.kt:182)
	at reactor.core.publisher.FluxSource.subscribe(FluxSource.java:67)
	at reactor.core.publisher.Flux.subscribe(Flux.java:8660)
	at reactor.core.publisher.FluxUsingWhen$ResourceSubscriber.onNext(FluxUsingWhen.java:195)
	at reactor.core.publisher.Operators$BaseFluxToMonoOperator.completePossiblyEmpty(Operators.java:2034)
	at reactor.core.publisher.MonoHasElements$HasElementsSubscriber.onComplete(MonoHasElements.java:93)
	at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89)
	at io.r2dbc.postgresql.util.FluxDiscardOnCancel$FluxDiscardOnCancelSubscriber.onComplete(FluxDiscardOnCancel.java:104)
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
	at reactor.core.publisher.FluxHandle$HandleSubscriber.onComplete(FluxHandle.java:222)
	at io.r2dbc.postgresql.util.FluxDiscardOnCancel$FluxDiscardOnCancelSubscriber.onComplete(FluxDiscardOnCancel.java:104)
	at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onComplete(FluxContextWrite.java:126)
	at reactor.core.publisher.FluxCreate$BaseSink.complete(FluxCreate.java:460)
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:805)
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.complete(FluxCreate.java:753)
	at reactor.core.publisher.FluxCreate$SerializedFluxSink.drainLoop(FluxCreate.java:247)
	at reactor.core.publisher.FluxCreate$SerializedFluxSink.drain(FluxCreate.java:213)
	at reactor.core.publisher.FluxCreate$SerializedFluxSink.complete(FluxCreate.java:204)
	at io.r2dbc.postgresql.client.ReactorNettyClient$Conversation.complete(ReactorNettyClient.java:671)
	at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.emit(ReactorNettyClient.java:937)
	at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.onNext(ReactorNettyClient.java:813)
	at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.onNext(ReactorNettyClient.java:719)
	at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:128)
	at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
	at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:224)
	at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:224)
	at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:292)
	at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:401)
	at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:411)
	at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:113)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:800)
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:499)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:397)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:833)

Environment Information

• Operating System: Ubuntu 22.04.1 LTS
• Micronaut Version: TODO
• JDK Version: openjdk 17.0.5 2022-10-18

Example Application

Here you can find some tests that reproduces the error

• https://github.com/filipenfst/micronaut-data/blob/master/doc-examples/r2dbc-example-kotlin/src/test/kotlin/example/TxTest3.kt

Start with doc-examples/r2dbc-example-kotlin/src/test/kotlin/example/TxTest3.kt, which reproduces the failure, and trace the transaction handling shown in DefaultR2dbcRepositoryOperations.java around the reported stack frame. Run the example tests and verify that coroutine and reactive Flow operations propagate the transaction, avoid NoTransactionException, and commit after the flow is collected.