Reactive Client ResponseExceptionMapper throws BlockingNotAllowedException when consuming an InputStream

[ssaip]

Describe the bug

When trying to read the response entity in an ResponseExceptionMapper, a BlockingNotAllowedException is thrown IF the return type of the call is an InputStream. It works fine with String.

My Application subclass is annotated @Blocking and I can also do blocking operations like a Thread.sleep in the ExceptionMapper

Expected behavior

InputStreams can be handled

Actual behavior

Using response.readEntity(JsonNode.class) in a ResponseExceptionMapper throws a BlockingNotAllowedException when the response type is InputStream

How to Reproduce?

see https://github.com/ssaip/quarkus-reactive-client-inputstream-blocking

org.jboss.resteasy.reactive.common.core.BlockingNotAllowedException: Attempting a blocking read on io thread
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream$VertxBlockingInput.readBlocking(VertxClientInputStream.java:199)
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream.readIntoBuffer(VertxClientInputStream.java:81)
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream.read(VertxClientInputStream.java:62)
	at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.ensureLoaded(ByteSourceJsonBootstrapper.java:539)
	at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.detectEncoding(ByteSourceJsonBootstrapper.java:133)
	at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.constructParser(ByteSourceJsonBootstrapper.java:256)
	at com.fasterxml.jackson.core.JsonFactory._createParser(JsonFactory.java:1685)
	at com.fasterxml.jackson.core.JsonFactory.createParser(JsonFactory.java:1084)
	at com.fasterxml.jackson.databind.ObjectReader.createParser(ObjectReader.java:1103)
	at com.fasterxml.jackson.databind.ObjectReader.readValue(ObjectReader.java:1481)
	at io.quarkus.resteasy.reactive.jackson.runtime.serialisers.ServerJacksonMessageBodyReader.doReadFrom(ServerJacksonMessageBodyReader.java:87)
	at io.quarkus.resteasy.reactive.jackson.runtime.serialisers.ServerJacksonMessageBodyReader.readFrom(ServerJacksonMessageBodyReader.java:38)
	at org.jboss.resteasy.reactive.client.impl.ClientReaderInterceptorContextImpl.proceed(ClientReaderInterceptorContextImpl.java:67)
	at org.jboss.resteasy.reactive.client.impl.ClientSerialisers.invokeClientReader(ClientSerialisers.java:160)
	at org.jboss.resteasy.reactive.client.impl.ClientResponseImpl.readEntity(ClientResponseImpl.java:59)
	at org.jboss.resteasy.reactive.common.jaxrs.ResponseImpl.readEntity(ResponseImpl.java:129)
	at com.example.BlockingResponseExceptionMapper.toThrowable(BlockingResponseExceptionMapper.java:24)
	at com.example.BlockingResponseExceptionMapper.toThrowable(BlockingResponseExceptionMapper.java:10)
	at com.example.BlockingResponseExceptionMapper_ClientProxy.toThrowable(Unknown Source)
	at io.quarkus.rest.client.reactive.runtime.MicroProfileRestClientResponseFilter.filter(MicroProfileRestClientResponseFilter.java:36)
	at org.jboss.resteasy.reactive.client.handlers.ClientResponseFilterRestHandler.handle(ClientResponseFilterRestHandler.java:21)
	at org.jboss.resteasy.reactive.client.handlers.ClientResponseFilterRestHandler.handle(ClientResponseFilterRestHandler.java:10)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.invokeHandler(AbstractResteasyReactiveContext.java:229)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:145)
	at org.jboss.resteasy.reactive.client.impl.RestClientRequestContext$1.lambda$execute$0(RestClientRequestContext.java:279)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:246)
	at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:43)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:829)

Output of uname -a or ver

No response

Output of java -version

Java version: 11.0.17, vendor: Azul Systems, Inc.

GraalVM version (if different from Java)

No response

Quarkus version or git rev

2.15.3.Final

Build tool (ie. output of mvnw --version or gradlew --version)

apache-maven-3.8.6

Additional information

No response

[quarkus-bot]

/cc @Sgitario (rest-client), @cescoffier (rest-client)

[ssaip]

Hi @Sgitario , @cescoffier just a heads-up: I updated the sample project to 2.16.1.Final, unfortunately the problem still exists.

[kanha-gupta]

@ssaip can you provide some direction ? I am a newcomer & ready to work :)

[cescoffier]

We didn't change anything, so, for sure, the bug is still there.

Unfortunately, I don't see a good approach to fix it, except requiring an @Blocking annotation (or a blocking mapper interface) and delegating the invocation of this mapper on a worker thread.

@Sgitario @geoand WDYT?

[geoand]

Exactly, no good ways around this one...

[ssaip]

Identifying the @Blocking annotation on the Mapper itself (or Application subclass ofc) and delegating to a worker thread sounds good to me.

[geoand]

I'll have a look at this tomorrow and see what can be done.

[geoand]

As is mentioned here, this only becomes a problem when InputStream is used as the return type of the REST Client method, so I don't think it makes sense to address the issue with support for @Blocking.

In any case we could (by doing plenty of additional work) run the exception handler on a worker thread, but I am wondering whether that would be too susprising...

In short I am not sure what the proper way to handle this is, especially considering it's an edge case.

[ssaip]

agreed this may be an edge-case and a lot of work needs to be done here, but in our case this is the sole reason we cannot migrate to the reactive client. What would you advise us?

[geoand]

May I ask why you need to return InputStream from your REST Client method?

[nikosk686]

Same issue affects a sligtly more common case. That is the need to parse the request body from within a RequestFilter (related open ticket here).
So the follwing code is also affected from the same issue.

@Blocking
@Path("/hello")
public class ExampleResource {

    @ServerRequestFilter(preMatching = true)
    public void filter(ContainerRequestContext requestContext) {
        String body = new BufferedReader(new InputStreamReader(requestContext.getEntityStream()))
                .lines().collect(Collectors.joining("\n"));
        requestContext.setEntityStream(new ByteArrayInputStream(body.getBytes()));
    }


    @POST
    @Produces(MediaType.TEXT_PLAIN)
    @Consumes(MediaType.APPLICATION_JSON)
    public Uni<String> hello(Map<String, String> body) throws InterruptedException {
        return Uni.createFrom().item("Hello from %s".formatted(body.get("name")));
    }
}

[geoand]

@nikosk686 that is for the server part, no?

[nikosk686]

@geoand this is for the server part indeed, not the rest client. However the exception is the same and it looks like @Blocking is being ignored when we try to parse the InputStream.

[geoand]

Okay. Asking because this issue is specific to the client.

Please open a new issue and attach a sample application that exhibits the problem you are seeing in the server which although looks similar, has a different root cause and is likely easy to fix).

[forsti0506]

Are there any plans to fix that or any solution I can apply in my project?

May I ask why you need to return InputStream from your REST Client method?

In our case it is used to transfer large files

[geoand]

In our case it is used to transfer large files

@Sgitario what is our recommendation for this use case?

[Sgitario]

In our case it is used to transfer large files

@Sgitario what is our recommendation for this use case?

Unfortunately, I could not find any workaround to read the InputStream from exception mappers.
About the issue, this seems to be caused by this line:

Context context = restClient.getVertx().getOrCreateContext();
            return new Executor() {
                @Override
                public void execute(Runnable command) {
                    context.runOnContext(v -> command.run());
                }
            };

I'm not very familiar with the code, but at this point, the context is always using the Event Loop Thread and hence it's being propagated to the REST Client chain (the exception mapper filters). Yet I think the intention was to reuse the caller context (but using @Blocking in the caller makes no effect).

I think the better solution would be to allow the usage of the @Blocking annotation in exception mappers, though I think this can be quite complex (and actually, I don't still know why it does not work).

Another simpler and much uglier solution is to use the blocking context if the returning type is an InputStream:

Context context = restClient.getVertx().getOrCreateContext();
            return new Executor() {
                @Override
                public void execute(Runnable command) {
                    if (responseTypeSpecified && responseType.getRawType().equals(InputStream.class)) {
                        context.executeBlocking(v -> command.run());
                    } else {
                        context.runOnContext(v -> command.run());
                    }
                }
            };

I tried this and everything works fine.

And of course, another solution is not to use the VertxInputStream in the REST Client, but a version that does not throw this exception (maybe logging a warning instead).

wdyt? @geoand

[geoand]

I think the better solution would be to allow the usage of the @Blocking annotation in exception mappers, though I think this can be quite complex (and actually, I don't still know why it does not work).

Yeah, I think this is the most reasonable way to go.

It is likely a complex underaking, because currently everything in the internals of the REST Client is run on the event loop.

[Sgitario]

I think the better solution would be to allow the usage of the @Blocking annotation in exception mappers, though I think this can be quite complex (and actually, I don't still know why it does not work).

Yeah, I think this is the most reasonable way to go.

It is likely a complex underaking, because currently everything in the internals of the REST Client is run on the event loop.

I would have expected that the @Blocking annotation handles exactly this situation. Any idea why it does not work / or any pointers about how we could support it?

[geoand]

The annotation is never read for these client exception mapper classes.

ResponseExceptionMapper classes are handled as part of MicroProfileRestClientResponseFilter, which in turn is handled by ClientResponseFilterRestHandler (which is always run on an event loop)

[Sgitario]

The annotation is never read for these client exception mapper classes.

ResponseExceptionMapper classes are handled as part of MicroProfileRestClientResponseFilter, which in turn is handled by ClientResponseFilterRestHandler (which is always run on an event loop)

So, the solution is that the ClientResponseFilterRestHandler handlers should check for the @Blocking annotation and if it's present, do not execute the handler on an event loop.

[geoand]

Well, we certainly don't want to check annotations at runtime, but in general, yes, that should likel work

[geoand]

However, there is an extra complication here...

Currently all ResponseExceptionMapper are run by the same MicroProfileRestClientResponseFilter, but don't want one mapper changing the behavior for all

[ssaip]

I've finally had the chance to migrate to 3.2.5.Final but I'm still getting the exact same error.

Updated example project: https://github.com/ssaip/quarkus-reactive-client-inputstream-blocking

What am I doing wrong??

Stacktrace:

org.jboss.resteasy.reactive.common.core.BlockingNotAllowedException: Attempting a blocking read on io thread
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream$VertxBlockingInput.readBlocking(VertxClientInputStream.java:199)
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream.readIntoBuffer(VertxClientInputStream.java:81)
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream.read(VertxClientInputStream.java:62)
	at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.ensureLoaded(ByteSourceJsonBootstrapper.java:539)
	at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.detectEncoding(ByteSourceJsonBootstrapper.java:133)
	at com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.constructParser(ByteSourceJsonBootstrapper.java:256)
	at com.fasterxml.jackson.core.JsonFactory._createParser(JsonFactory.java:1744)
	at com.fasterxml.jackson.core.JsonFactory.createParser(JsonFactory.java:1143)
	at com.fasterxml.jackson.databind.ObjectReader.createParser(ObjectReader.java:1103)
	at com.fasterxml.jackson.databind.ObjectReader.readValue(ObjectReader.java:1481)
	at io.quarkus.rest.client.reactive.jackson.runtime.serialisers.ClientJacksonMessageBodyReader.readFrom(ClientJacksonMessageBodyReader.java:53)
	at org.jboss.resteasy.reactive.client.impl.ClientReaderInterceptorContextImpl.proceed(ClientReaderInterceptorContextImpl.java:86)
	at org.jboss.resteasy.reactive.client.impl.ClientSerialisers.invokeClientReader(ClientSerialisers.java:160)
	at org.jboss.resteasy.reactive.client.impl.ClientResponseImpl.readEntity(ClientResponseImpl.java:59)
	at org.jboss.resteasy.reactive.common.jaxrs.ResponseImpl.readEntity(ResponseImpl.java:129)
	at com.example.BlockingResponseExceptionMapper.toThrowable(BlockingResponseExceptionMapper.java:17)
	at com.example.BlockingResponseExceptionMapper.toThrowable(BlockingResponseExceptionMapper.java:10)
	at com.example.BlockingResponseExceptionMapper_ClientProxy.toThrowable(Unknown Source)
	at io.quarkus.rest.client.reactive.runtime.MicroProfileRestClientResponseFilter.filter(MicroProfileRestClientResponseFilter.java:52)
	at org.jboss.resteasy.reactive.client.handlers.ClientResponseFilterRestHandler.handle(ClientResponseFilterRestHandler.java:21)
	at org.jboss.resteasy.reactive.client.handlers.ClientResponseFilterRestHandler.handle(ClientResponseFilterRestHandler.java:10)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.invokeHandler(AbstractResteasyReactiveContext.java:229)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:145)
	at org.jboss.resteasy.reactive.client.impl.RestClientRequestContext$1.lambda$execute$0(RestClientRequestContext.java:302)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:246)
	at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:43)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:829)

/cc @Sgitario, @cescoffier, @geoand

[geoand]

@Sgitario any quick ideas?

[Sgitario]

This is caused because the exception mapper is a bean (it's using the @ApplicationScoped annotation).
This can be easily fixed. Tho, @ssaip if you replace @ApplicationScoped with @Provider as noted in the guide, it will work fine.

[geoand]

Ah, interesting. So it's an invalid usage then.

Thanks!

[ssaip]

Thank you so much @Sgitario, replacing the @ApplicationScoped with @Provider fixed it.

[ssaip]

While this was working in my sample project, I still couldn't get it to work in the real project.
I'll keep this here and maybe it saves someone else some hours in the future.

I figured out that my ResponseExceptionMapper always got _subclass'd because I was using jakarta.validation annotations there and had the io.quarkus:quarkus-hibernate-validator dependency on the classpath, therefore this class was always getting proxied to intercept for validations.

And because of that, the @Blocking annotation was no longer present and the ResponseExceptionMapper was run on the io thread.
Unfortunately, there is no way to fix the annotations for sub/proxied classes (a topic that I originally came across when investigating the cause of this issue :D see #30327)

Anyway, removing annotations like @NotNull, will prevent subclassing and the @Blocking ResponseExceptionMapper will run on a worker-thread as expected.