Reactive Transactions in Kotlin: @Transactional, TransactionalOperator, and the Coroutine Context

Classic Spring transactions are bound to a thread. The framework stashes the JDBC connection in a `ThreadLocal`, and as long as your code stays on that thread every repository call quietly enlists in the same transaction. Reactive Spring throws that model away. With R2DBC there is no thread affinity at all: a single request can hop across many worker threads as it suspends and resumes, so a `ThreadLocal` would be useless. Instead, the active transaction is carried inside the Reactor `Context` (or, for coroutines, the `CoroutineContext`) that travels with the reactive subscription. Understanding this single shift is the key to everything else in this lesson.

Spring bridges this for you through `org.springframework.transaction.reactive.TransactionContext`, and the good news is that `@Transactional` still works on `suspend` functions. When you annotate a suspend function (or a function returning `Mono`/`Flux`), Spring's `TransactionInterceptor` opens a reactive transaction, propagates its context through the coroutine, and commits when the returned publisher completes successfully or rolls back when it fails. The crucial requirement is that the bean must be a Spring-managed component and the call must come from outside the bean, because the interceptor lives in a proxy. Self-invocation (calling another `@Transactional` method via `this`) bypasses the proxy and silently runs with no transaction at all.

For finer-grained, programmatic control you reach for `TransactionalOperator`. It wraps a piece of reactive work so that everything subscribed inside it shares one transaction, and it exposes a coroutine-friendly extension, `executeAndAwait { ... }`, that lets you write straight-line suspending code. This is ideal when you only want part of a method to be transactional, when you need to commit conditionally, or when annotations feel too coarse. Because the operator manages the boundary explicitly, it composes cleanly with `coroutineScope` and structured concurrency, and it makes the transactional region obvious at the call site rather than hidden behind an annotation.

Propagation behaves like its blocking cousin but with a reactive twist. `REQUIRED` (the default) joins an existing transaction if the context already carries one, or starts a new one otherwise. `REQUIRES_NEW` suspends the outer transaction and runs in a brand-new one, which is perfect for an audit-log write that must persist even if the main operation rolls back. Be aware that R2DBC's reactive transaction manager does not support every propagation mode that JPA does (for example, `NESTED` savepoint behavior is not supported), so check your driver. The mental model to keep is: propagation decides whether a new `TransactionContext` is created or an existing one in the subscriber context is reused.

Now the headline danger: blocking inside a reactive transaction. Reactor and coroutine pipelines run on a tiny, fixed pool of non-blocking event-loop threads. If you call a blocking JDBC repository, `Thread.sleep`, a synchronous `RestTemplate`, or any `.block()`/`runBlocking` inside a transactional reactive chain, you park one of those precious threads. Under load this starves the event loop, latency spikes, throughput collapses, and requests time out. Worse, the blocking work is not part of the reactive transaction's context, so a blocking JDBC call executes on its own connection with its own (or no) transaction. Your rollback will never undo it, quietly breaking atomicity.

The fix is to keep the entire transactional path non-blocking. Use R2DBC repositories, reactive `DatabaseClient`, and `WebClient` instead of their blocking equivalents, and never call `.awaitSingle()` on a blocking source. If you genuinely must invoke legacy blocking code, push it off the event loop with `withContext(Dispatchers.IO)` so you do not starve the reactive scheduler, but understand clearly that the blocking work then runs outside the reactive transaction and will not be rolled back with it. The only way to keep something inside the transaction is to perform it through a reactive, R2DBC-aware client that participates in the same `TransactionContext`.

Rollback semantics deserve a note. In the reactive world a transaction rolls back when the returned publisher or suspend function terminates with an error. That means you must let exceptions propagate; if you swallow an error with a `try/catch` and return normally, Spring sees success and commits. Likewise, cancellation matters: if the coroutine scope is cancelled mid-transaction, the reactive transaction is rolled back, which is usually what you want but can surprise teams migrating from blocking code where cancellation was rare.

A practical checklist closes the loop. Put `@Transactional` on a public method of a Spring bean and call it from another bean; prefer R2DBC end to end; reach for `TransactionalOperator.executeAndAwait` when you need explicit boundaries; use `REQUIRES_NEW` for must-survive side effects like audit logs; and audit your dependency graph for any blocking driver hiding behind a reactive interface. If you remember only one sentence, make it this: in reactive Spring the transaction lives in the context that flows with your coroutines, and the moment you block or step outside that context, the guarantee is gone.

import org.springframework.stereotype.Service
import org.springframework.transaction.annotation.Propagation
import org.springframework.transaction.annotation.Transactional

@Service
class OrderService(
    private val orders: OrderRepository,      // R2DBC CoroutineCrudRepository
    private val auditService: AuditService,
) {
    // The interceptor opens a reactive tx, propagates it through the
    // coroutine context, and commits when the function returns normally.
    @Transactional
    suspend fun placeOrder(order: Order): Order {
        val saved = orders.save(order)            // joins the same tx
        auditService.record("order_placed", saved.id) // see REQUIRES_NEW below
        if (saved.total < 0) error("invalid total") // throwing -> rollback
        return saved
    }
}

@Service
class AuditService(private val audit: AuditRepository) {
    // Runs in its own brand-new tx, so the audit row survives even if
    // placeOrder later rolls back. Must be called from another bean.
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    suspend fun record(event: String, id: Long?) {
        audit.save(AuditEntry(event = event, refId = id))
    }
}

import org.springframework.stereotype.Service
import org.springframework.transaction.ReactiveTransactionManager
import org.springframework.transaction.reactive.TransactionalOperator
import org.springframework.transaction.reactive.executeAndAwait

@Service
class TransferService(
    private val accounts: AccountRepository,   // R2DBC, fully reactive
    rtm: ReactiveTransactionManager,           // auto-configured by Spring Boot
) {
    // Spring Boot auto-configures a ReactiveTransactionManager for R2DBC,
    // but NOT a TransactionalOperator — build one from the manager.
    private val tx = TransactionalOperator.create(rtm)

    suspend fun transfer(fromId: Long, toId: Long, amount: Long) =
        // Everything inside shares ONE reactive transaction. If the block
        // throws (or the coroutine is cancelled), the whole thing rolls back.
        tx.executeAndAwait {
            val from = accounts.findById(fromId) ?: error("no source account")
            val to = accounts.findById(toId) ?: error("no target account")
            require(from.balance >= amount) { "insufficient funds" }
            accounts.save(from.copy(balance = from.balance - amount))
            accounts.save(to.copy(balance = to.balance + amount))
        }
}

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext

// ANTI-PATTERN: a blocking call on a reactive event-loop thread.
// It starves the scheduler AND runs outside the reactive transaction,
// so legacyJdbc.write(...) will NOT be rolled back with the tx.
@Transactional
suspend fun broken(order: Order) {
    orders.save(order)              // reactive, in the tx
    legacyJdbc.write(order)         // BLOCKS the event loop, escapes the tx
}

// If you must call blocking code, at least move it off the event loop.
// Still outside the reactive tx — accept that and design around it.
@Transactional
suspend fun lessBad(order: Order) {
    orders.save(order)
    withContext(Dispatchers.IO) { legacyJdbc.write(order) } // no longer starves
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking

// Pure stdlib + coroutines demo (no Spring/DB): a non-blocking 'delay'
// suspends without holding a thread, while a blocking sleep would not.
fun main() = runBlocking {
    println("start")
    delay(100)               // suspends, frees the thread (non-blocking)
    // Thread.sleep(100)     // would block the thread — the reactive sin
    println("resumed without parking a thread")
}