Seeking advice on running blocking code and context/thread management?

[gunjanthesystem]

Aim: to execute 3 tasks in sequence: consider them like one time thing to do when the application starts - like data cleaning, transformation and load. Below is a toy example to demonstrate:

Task 1: has calls to blocking methods like loading data from GCP buckets.
Task 2: is non blocking data transformation
Task 3: is loading data into postgres db using Hibernate Reactive Panache.

Question 1: What is the recommended way to execute task1WithBlocking on worker thread. using runSubscriptionOn or there is better approach including design change.

Question 2: When task1WithBlocking is put on executor thread, how to keep Task 2 and Task 3 on the original thread?

Question 3: OnStart starts on Quarkus Main Thread, this will cause issues for Hibernate Reactive Panache as it need vertx event loop. Is there a better way?

Please advise if there is a better way to do this given the contraints. (Trying to shift to a reactive approach but stuck on this issue for a while. Open to all suggestions.)

import io.quarkus.runtime.StartupEvent;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Observes;
import org.jboss.logging.Logger;

@ApplicationScoped
public class QuarkusBlockingExample
{
    private static final Logger logger = Logger
            .getLogger(QuarkusBlockingExample.class);

    void onStart(@Observes StartupEvent ev)
    {
        logger.info("Log 1: QuarkusBlockingExample onStart is starting...");

        executeDataPipeline().subscribe().with(
                success -> logger.info("Log 5: executeDataPipeline completed."),
                failure -> logger.error("Log 5f: executeDataPipeline failed",
                        failure));
        logger.info("Log 8: QuarkusBlockingExample onStart is finished.");
    }

    public Uni<Void> executeDataPipeline()
    {
        logger.info("Log 2: Starting executeDataPipeline");

        return Uni.createFrom().voidItem().onItem()
                .transformToUni(unused -> task1WithBlocking()).onItem()
                .transformToUni(unused -> task2NonBlocking()).onItem()
                .transformToUni(unused -> task3NonBlockingDB()).onItem()
                .invoke(() -> logger
                        .info("Log 7: Processing initialisation complete."));
    }

    public Uni<Void> task1WithBlocking()
    {
        return Uni.createFrom().voidItem().invoke(this::blockingMethod)
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .invoke(() -> logger
                        .info("Log 4: Completed task1WithBlocking."))
                .replaceWithVoid();
    }

    public Uni<Void> task2NonBlocking()
    {
        return Uni.createFrom().voidItem().invoke(
                () -> logger.info("Log 5: Completed task2NonBlocking."));
    }

    public Uni<Void> task3NonBlockingDB()
    {
        return Uni.createFrom().voidItem().invoke(() -> logger.info(
                "Log 6: Completed DB load using Reactive Hibernate Panache."));
    }

    private void blockingMethod()
    {
        try
        {
            Thread.sleep(2000);
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        logger.info("Log 3: Blocking method executed.");
    }
}

[cescoffier]

\CC @jponge

[jponge]

This is a cross-post of https://stackoverflow.com/questions/79391383/quarkus-with-mutiny-seeking-advice-on-running-blocking-code-and-context-thread which I intend to get back to later.