Implementing Async Soap Client - IO Thread vs worker thread

[piba84-lx]

Hi everyone,

I'm writing here to ask for help / suggestion on the usage of the CXF client in the reactive world.

Thanks to the example provided here: Issue #4 Comment, I was able to implement a reactive endpoint on my webservice that's "translating" the result of a future into a Uni.

I've noticed though that the number of threads spawned in the process of performing the SOAP request, parsing the response and provide it back the original caller is a bit different than what I was expecting. Specifically, I was expecting a mix of vertex thread and async client thread.
But what I'm seeing is an additional thread the executor-thread. Is there any doc that can explain me a bit better what's happening.

2022-12-29 18:25:14,039 INFO  [org.acm.AddNumbersAsync] (vert.x-eventloop-thread-1) I'm on the resource - Thread Name: vert.x-eventloop-thread-1
2022-12-29 18:25:14,043 INFO  [org.acm.ThreadLoggerInterceptor] (vert.x-eventloop-thread-1) Conduit: org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit, Thread Name: vert.x-eventloop-thread-1
2022-12-29 18:25:14,386 INFO  [org.acm.ThreadLoggerInterceptor] (default-workqueue-2) Conduit: org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit, Thread Name: default-workqueue-2
2022-12-29 18:25:14,388 INFO  [org.acm.AddNumbersAsync] (default-workqueue-2) I'm on the async handler - Thread Name: default-workqueue-2
2022-12-29 18:25:14,388 INFO  [org.acm.AddNumbersAsync] (executor-thread-0) I'm on the uni map function - Thread Name: executor-thread-0

I've slightly modified the code provided to mimic my code

@GET
    @Path("/{num1}/{num2}")
    public Uni<Integer> addNumbers(Integer num1, Integer num2) {
        LOGGER.infof("I'm on the resource - Thread Name: %s", Thread.currentThread().getName());
        return Uni.createFrom().future(() -> (Future<AddResponse>) calculatorSoap.addAsync(num1, num2, res -> {
            LOGGER.infof("I'm on the async handler - Thread Name: %s", Thread.currentThread().getName());
        })).map(response -> {
            LOGGER.infof("I'm on the uni map function - Thread Name: %s", Thread.currentThread().getName());
            return response.getAddResult();
        });
    }  

Thank you in advance for any link/doc/suggestion you might be able to provide.

Best,
Paolo

[shumonsharif]

Hi Paolo, I'm curious what happens when you use the same pattern without quarkus-cxf in the picture, but that still uses
a Future?

Also, not sure if it's helpful, but I believe you may be able to set the executor for the CXF service using something like:

ClientProxy.getClient(calculatorSoap).setExecutor(Infrastructure.getDefaultExecutor());

I'm suspecting this has more to do with Futures in general, as opposed to the extension -- I unfortunately don't have a good explanation, but kindly keep us posted on your findings!

[piba84-lx]

I’ll give it a try and see what happens.

[piba84-lx]

I've tried to apply the Infra default executor, but I'm getting the following results.

2022-12-30 23:12:27,234 INFO  [org.acm.AddNumbersAsync] (vert.x-eventloop-thread-0) I'm on the resource - Thread Name: vert.x-eventloop-thread-0
2022-12-30 23:12:27,238 INFO  [org.acm.ThreadLoggerInterceptor] (vert.x-eventloop-thread-0) Conduit: org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit, Thread Name: vert.x-eventloop-thread-0
2022-12-30 23:12:27,577 INFO  [org.acm.ThreadLoggerInterceptor] (executor-thread-1) Conduit: org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit, Thread Name: executor-thread-1
2022-12-30 23:12:27,578 INFO  [org.acm.AddNumbersAsync] (executor-thread-1) I'm on the async handler - Thread Name: executor-thread-1
2022-12-30 23:12:27,578 INFO  [org.acm.AddNumbersAsync] (executor-thread-0) I'm on the uni map function - Thread Name: executor-thread-0

I think it's now using the default thread pool for workloads, so I'll explore more following this lead.

thx for the help

[piba84-lx]

I was looking through the code and I noticed this comment:

// File: UniCreateFromFuture

private void dispatchDeferredResult(Future<? extends T> future, UniSubscriber<? super T> downstream) {
        AtomicBoolean cancelled = new AtomicBoolean(false);
        downstream.onSubscribe(() -> {
            cancelled.set(true);
            future.cancel(false);
        });
        // Because future.get is blocking, we must use a separated thread. <-- THIS COMMENT SEEMS VERY CLEAR
        Infrastructure.getDefaultExecutor().execute(() -> { ...

so forcing the executor is crashing the app:

 io.vertx.core.VertxException: Thread blocked
        at java.base@11.0.1/jdk.internal.misc.Unsafe.park(Native Method)
        at java.base@11.0.1/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)
        at java.base@11.0.1/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1796)
        at java.base@11.0.1/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3128)
        at java.base@11.0.1/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1823)
        at java.base@11.0.1/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1998)
        at org.apache.cxf.endpoint.ClientCallback.get(ClientCallback.java:139)
        at org.apache.cxf.jaxws.JaxwsResponseCallback.get(JaxwsResponseCallback.java:49)
        at io.smallrye.mutiny.operators.uni.builders.UniCreateFromFuture.lambda$dispatchDeferredResult$1(UniCreateFromFuture.java:81)
        at io.smallrye.mutiny.operators.uni.builders.UniCreateFromFuture$$Lambda$1048/0x0000000800915840.run(Unknown Source)
        at io.smallrye.faulttolerance.vertx.VertxExecutor.execute(VertxExecutor.java:21)
        at io.smallrye.mutiny.operators.uni.builders.UniCreateFromFuture.dispatchDeferredResult(UniCreateFromFuture.java:79)
        at io.smallrye.mutiny.operators.uni.builders.UniCreateFromFuture.subscribe(UniCreateFromFuture.java:31)
        at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
        at io.smallrye.mutiny.operators.uni.UniOnItemTransform.subscribe(UniOnItemTransform.java:22)
        at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
        at io.smallrye.mutiny.groups.UniSubscribe.withSubscriber(UniSubscribe.java:52)