JOOQ 3.17.2 - Attempt to execute a blocking method when only an R2BDC ConnectionFactory was configured

user_7915834 0
Viewed 107

Dependencies used:

    implementation("org.jooq:jooq:3.17.2")
    implementation("org.jooq:jooq-meta:3.17.2")
    implementation("org.jooq:jooq-kotlin-coroutines:3.17.2")
    implementation("org.jooq:jooq-kotlin:3.17.2")
    runtimeOnly("io.r2dbc:r2dbc-spi") {
        version { strictly("0.9.1.RELEASE") }
    }
    runtimeOnly("dev.miku:r2dbc-mysql:0.8.2.RELEASE")
    runtimeOnly("io.r2dbc:r2dbc-pool:0.9.1.RELEASE")

     // Kotlin 1.6.20
    implementation("io.projectreactor.kotlin:reactor-kotlin-extensions")
    implementation("org.jetbrains.kotlin:kotlin-reflect")
    implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-jdk8")

Using R2DBC DSL configuration:

private val dsl = DSL.using(
        ConnectionFactories.get(
            ConnectionFactoryOptions
                .parse("r2dbc:pool:mysql://localhost:3303/db")
                .mutate()
                .option(ConnectionFactoryOptions.USER, databaseUser)
                .option(ConnectionFactoryOptions.PASSWORD, databasePassword)
                .build()
        )
    )

Running on JDK17:

DSL.using(dsl.configuration()).insertInto(table)
                .set(fromDomain(item)).apply(builder)
                .executeAsync()
                .await()

throws an error on line with .await()

Attempt to execute a blocking method (e.g. Query.execute() or ResultQuery.fetch()) when only an R2BDC ConnectionFactory was configured. jOOQ's RowCountQuery and ResultQuery extend Publisher, which allows for reactive streams implementations to subscribe to the results of a jOOQ query. Simply embed your query in the stream, e.g. using Flux.from(query). See also: https://www.jooq.org/doc/latest/manual/sql-execution/fetching/reactive-fetching/
org.jooq.exception.DetachedException: Attempt to execute a blocking method (e.g. Query.execute() or ResultQuery.fetch()) when only an R2BDC ConnectionFactory was configured. jOOQ's RowCountQuery and ResultQuery extend Publisher, which allows for reactive streams implementations to subscribe to the results of a jOOQ query. Simply embed your query in the stream, e.g. using Flux.from(query). See also: https://www.jooq.org/doc/latest/manual/sql-execution/fetching/reactive-fetching/
    (Coroutine boundary)
    at ...
Caused by: org.jooq.exception.DetachedException: Attempt to execute a blocking method (e.g. Query.execute() or ResultQuery.fetch()) when only an R2BDC ConnectionFactory was configured. jOOQ's RowCountQuery and ResultQuery extend Publisher, which allows for reactive streams implementations to subscribe to the results of a jOOQ query. Simply embed your query in the stream, e.g. using Flux.from(query). See also: https://www.jooq.org/doc/latest/manual/sql-execution/fetching/reactive-fetching/
    at org.jooq_3.17.2.MYSQL.debug(Unknown Source)
    at app//org.jooq.impl.AbstractQuery.execute(AbstractQuery.java:300)
    at app//org.jooq.impl.Tools$3$1.block(Tools.java:5803)
    at java.base@17.0.1/java.util.concurrent.ForkJoinPool.compensatedBlock(ForkJoinPool.java:3449)
    at java.base@17.0.1/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3432)
    at app//org.jooq.impl.Tools$3.get(Tools.java:5800)
    at java.base@17.0.1/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768)
    at java.base@17.0.1/java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1760)
    at java.base@17.0.1/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373)
    at java.base@17.0.1/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182)
    at java.base@17.0.1/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655)
    at java.base@17.0.1/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622)
    at java.base@17.0.1/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165)

although executeAsync is used instead of execute. I would not expect this error and would expect the query to run successfully. Thanks for any help in trying to solve this issue.

r2dbc jooq kotlin-coroutines kotlin mysql
2 Answers

I believe I understand it now. Instead of .executeAsync().await() I can use .awaitLast() and instead of fetchAsync() I can use .asFlow().toList()

user_79158340

There's a known limitation (as of jOOQ 3.17) that jOOQ doesn't execute fetchAsync() and other CompletionStage producing API calls via R2DBC yet, see: https://github.com/jOOQ/jOOQ/issues/13590

But since you're using kotlin coroutines as a "frontend" of jOOQ reactive queries, just use the recommended bridge module:

<dependency>
  <groupId>org.jooq</groupId>
  <artifactId>jooq-kotlin-coroutines</artifactId>
</dependency>

It allows you to use jOOQ like this, for example:

fun coroutinesWithTransactions() {
    val actor: ActorRecord = runBlocking {
        insertActorTransaction()
    }

    println(actor)
}

suspend fun insertActorTransaction(): ActorRecord {
    return ctx.transactionCoroutine(::insertActor)
}

suspend fun insertActor(c: Configuration): ActorRecord = c.dsl()
    .insertInto(ACTOR)
    .columns(ACTOR.ACTOR_ID, ACTOR.FIRST_NAME, ACTOR.LAST_NAME)
    .values(201L, "A", "A")
    .returning()
    .awaitFirst()

See also:

user_5217990
Related