How to propagte context to parallelStream operations in Quarkus?

user_1479480 0
Viewed 500

I have a simple series of chained operations that retrieve and persist some data using a Panache repository, running in a Quarkus service. Where these operations are parallelised a ContextNotActiveException is thrown. Where the parallelisation is removed, the code works as intended.

This code works:

    dataRepository.get()
        .map { convert(it) }
        .forEach { perist(it) }

This code does not:

    dataRepository.get()
        .parallelStream()
        .map { convert(it) }
        .forEach { perist(it) }

The Quarkus documentation is pretty limited, only addressing use of mutiny or RX.

How can I propagate the context such that parallelStream() will work?

quarkus
1 Answers

Unfortunately Context Propagation does not play well with parallel Java streams, because making a stream parallel automatically moves the execution to the ForkJoinPool, which means you lose the context. You'll need to handle the parallelism differently, without having the Java streams do it for you - you will probably want to use the org.eclipse.microprofile.context.ManagedExecutor.

Assuming that it's the convert method which, for whatever reason, requires an active request context, you will need to dispatch its invocation into the managed executor. This will make sure that the context is propagated. In Java code, one close equivalent to your code that I can think of is this:

    @Inject
    org.eclipse.microprofile.context.ManagedExecutor executor;

(...)

dataRepository.streamAll()
                .forEach(i -> {
                    executor.supplyAsync(() -> {
                        return convert(i);
                    }).thenAccept(persist(i));
                });
user_6151040
Related