Kotlin coroutines and R2DBC

Intro

This article demonstrates how to use reactive R2DBC drivers with Kotlin coroutines. To illustrate this, a typical outbox pattern implementation using Spring, Kotlin, PostgreSQL, and JDBC is reworked into a reactive form. Usage of Project Reactor is kept to a minimum to maintain a coroutines-first approach.

Why the hell do I need R2DBC?

• non-blocking DB access so threads aren’t blocked on IO
• higher concurrency with fewer threads
• fits reactive stacks (WebFlux, reactive messaging)
• useful for IO-heavy, high-load services

Why the hell do I need coroutines?

• Kotlin first-class feature with language-level support
• write async code in simple sequential style
• less cumbersome than Reactor’s Flux/Mono chains

Wait, can I refrain from Reactor at all?

Other Spring reactive components used for end-to-end reactive applications, such as WebFlux or reactive Kafka client, are built on top of Reactor. However, this does not mean that one must use Reactor in the application code. It is possible to write code with coroutines only (no Flux/Mono in app), and keep Reactor at the edges as an underlying runtime. In general, there is nothing wrong with mixing Reactor and coroutines or staying with coroutines only.

Preparation of components

Now Let’s look at the components involved.

R2DBC

R2DBC drivers are database drivers that implement reactive, non-blocking access to relational databases. In simple terms, they are the reactive alternative to JDBC drivers. They have been around since about 2019 and have been steadily gaining popularity since then. However, JDBC remains dominant in most applications because of its maturity, stability and simplicity. This demo also aims to show that R2DBC usage can be straightforward in practice.

We will use the PostgreSQL driver for this demo:

implementation("org.postgresql:r2dbc-postgresql")

Spring

Spring provides the basic reactive tools to work with R2DBC drivers via the

implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")

It also transitively includes the Project Reactor library, which provides the reactive types and runtime.

JOOQ

In this demo we jOOQ to interact with the database. jOOQ supports R2DBC drivers since v3.15

The spring-boot-starter-jooq starter is not compatible because it uses JDBC. So we have to add

implementation("org.jooq:jooq")
implementation("org.jooq:jooq-kotlin-coroutines")

A DSL Context object has to be configured and provided like

@Bean
fun dslContext(cf: io.r2dbc.spi.ConnectionFactory): org.jooq.DSLContext =
    DSL.using(cf, SQLDialect.POSTGRES)

Flyway

Flyway does not support R2DBC drivers and therefore has to be configured separately to have its own separate jdbc connection from configs like:

spring:
  flyway:
    url: jdbc:postgresql://...

This is fine because in Spring Boot, the Flyway datasource is separate from the application datasource by default. The Flyway datasource is used only to run migrations. The Flyway datasource does not interfere with the R2DBC setup.

Testcontainers

Testcontainers have to be configured to work with R2DBC:

testImplementation("org.testcontainers:testcontainers-r2dbc")

How do we adapt the code?

With the setup in place, let’s look at what changes in the code.

In reactive mode, a jOOQ Query returns org.reactivestreams.Publisher<Record>. Publisher is defined by Reactive Streams, a standalone, technology-agnostic spec for asynchronous stream processing. Project Reactor is just a popular Java implementation of this spec.

Kotlin coroutines use a different model with suspending functions and Flow type, which are conceptually similar to Mono and Flux types from Project Reactor. Jetbrains also provides a bridge library kotlinx-coroutines-reactive that enables interoperability between Publisher (for example, a Flux from a jOOQ query) and coroutines.

A normal insert query

fun insert(aggregateType: String, aggregateId: String, payload: String): Long {
    return dsl.insertInto(table("outbox"))
        ...
        .fetchSingle()

becomes a suspending function

suspend fun insert(aggregateType: String, aggregateId: String, payload: String): Long {
    return dsl.insertInto(table)
        ...
        .awaitFirst()

A select query which returns a list of records

fun selectUnpublished(limit: Int): List<OutboxRecord> {
    return dsl.selectFrom(table)
        .where(field("published_at").isNull())
        .orderBy(field("id"))
        .limit(limit)
        .fetch { deser(it) }
}

becomes either a suspending function

suspend fun selectUnpublished(limit: Int): List<OutboxRecord> {
    val query = dsl.selectFrom(table("outbox"))
        .where(field("published_at").isNull())
        .orderBy(field("id"))
        .limit(limit)
    return Flux.from(query)
        .map { deser(it) }
        .collectList()
        .awaitSingle()
}

or a function that returns a Flow

fun selectUnpublishedAsFlow(limit: Int): Flow<OutboxRecord> {
    val query = dsl.selectFrom(table("outbox"))
        .where(field("published_at").isNull())
        .orderBy(field("id"))
        .limit(limit)
    return Flux.from(query)
        .asFlow()
        .map { deser(it) }
}

Additional notes

Transaction management

Since Spring 5.3, the Spring @Transactional is aware of Kotlin coroutines. When a suspend function is marked @Transactional, Spring correctly manages the transaction context within the CoroutineContext. NB: @Transactional in tests is looking still for JDBC Data source and does not work correctly if it is not configured.

Scheduling

Starting with Spring 6.1, @Scheduled officially supports Kotlin suspend functions.

Note on Java virtual threads

Java virtual threads first appeared as a preview in Java 19 (Project Loom) and became stable in Java 21. They also make blocking code scale much better by making threads lightweight and inexpensive. In fact, virtual threads reduce the need for reactive style for scalability. However, they are still blocking from the driver perspective (JDBC blocks sockets).

In Spring Boot, virtual threads are disabled by default, but they can be enabled by setting spring.threads.virtual.enabled=true, allowing the framework to use them.

Summary

In this demo, we transformed a synchronous JDBC app into a reactive R2DBC one with relatively few changes and little effort.

Further steps may include moving the remaining blocking parts to non-blocking alternatives and building a fully end-to-end reactive application.

Links

• Repository with the demo code: https://github.com/fyrkov/r2dbc-coroutines-demo
• Reactive Streams specification: https://www.reactive-streams.org/
• Project Reactor https://projectreactor.io/docs
• Introduction of R2DBC to jOOQ: https://blog.jooq.org/reactive-sql-with-jooq-3-15-and-r2dbc/