Repository: quarkusio/quarkus
Issue: `IllegalStateException: The current thread cannot be blocked` using `quarkus-cache` with `quarkus-resteasy-reactive`

### Describe the bug

Using `quarkus-cache` together with reactive execution model causes an `IllegalStateException`.
Calling e.g. a reactive endpoint:
```java
@NonBlocking
@Path("/value")
public class ReactiveResource {

    public static final String CACHE_NAME = "mycache";

    private static int counter = 0;

    @CacheResult(cacheName = CACHE_NAME)
    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public Uni<String> getValue() {
        return Uni.createFrom().item("Value: " + counter++);
    }
}
```
results in:
```
java.lang.IllegalStateException: The current thread cannot be blocked: vert.x-eventloop-thread-14
```

### Expected behavior

OK response.

### Actual behavior

```
2021-12-15 16:26:55,525 ERROR [io.qua.ver.htt.run.QuarkusErrorHandler] (vert.x-eventloop-thread-14) HTTP Request to /value failed, error id: ec3dd35b-7d65-49bb-b718-8cddaeef8715-1: java.lang.IllegalStateException: The current thread cannot be blocked: vert.x-eventloop-thread-14
	at io.smallrye.mutiny.operators.uni.UniBlockingAwait.await(UniBlockingAwait.java:30)
	at io.smallrye.mutiny.groups.UniAwait.atMost(UniAwait.java:62)
	at io.smallrye.mutiny.groups.UniAwait.indefinitely(UniAwait.java:43)
	at io.quarkus.cache.runtime.CacheResultInterceptor.intercept(CacheResultInterceptor.java:65)
	at io.quarkus.cache.runtime.CacheResultInterceptor_Bean.intercept(Unknown Source)
	at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
	at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
	at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
	at org.acme.ReactiveResource_Subclass.getValue(Unknown Source)
	at org.acme.ReactiveResource$quarkusrestinvoker$getValue_8f786ee95084b5bc35ad81779e5d693d969db3a2.invoke(Unknown Source)
	at org.jboss.resteasy.reactive.server.handlers.InvocationHandler.handle(InvocationHandler.java:29)
	at org.jboss.resteasy.reactive.server.handlers.InvocationHandler.handle(InvocationHandler.java:7)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:141)
	at org.jboss.resteasy.reactive.server.handlers.RestInitialHandler.beginProcessing(RestInitialHandler.java:49)
	at org.jboss.resteasy.reactive.server.vertx.ResteasyReactiveVertxHandler.handle(ResteasyReactiveVertxHandler.java:17)
	at org.jboss.resteasy.reactive.server.vertx.ResteasyReactiveVertxHandler.handle(ResteasyReactiveVertxHandler.java:7)
	at io.vertx.ext.web.impl.RouteState.handleContext(RouteState.java:1193)
	at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:163)
	at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:141)
	at io.quarkus.vertx.http.runtime.VertxHttpRecorder$5.handle(VertxHttpRecorder.java:362)
	at io.quarkus.vertx.http.runtime.VertxHttpRecorder$5.handle(VertxHttpRecorder.java:340)
	at io.vertx.ext.web.impl.RouteState.handleContext(RouteState.java:1193)
	at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:163)
	at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:141)
	at io.vertx.ext.web.impl.RouterImpl.handle(RouterImpl.java:67)
	at io.vertx.ext.web.impl.RouterImpl.handle(RouterImpl.java:37)
	at io.quarkus.vertx.http.runtime.VertxHttpRecorder$12.handle(VertxHttpRecorder.java:486)
	at io.quarkus.vertx.http.runtime.VertxHttpRecorder$12.handle(VertxHttpRecorder.java:483)
	at io.quarkus.vertx.http.runtime.VertxHttpRecorder$1.handle(VertxHttpRecorder.java:152)
	at io.quarkus.vertx.http.runtime.VertxHttpRecorder$1.handle(VertxHttpRecorder.java:134)
	at io.vertx.core.http.impl.Http1xServerRequestHandler.handle(Http1xServerRequestHandler.java:67)
	at io.vertx.core.http.impl.Http1xServerRequestHandler.handle(Http1xServerRequestHandler.java:30)
	at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:50)
	at io.vertx.core.impl.DuplicatedContext.emit(DuplicatedContext.java:168)
	at io.vertx.core.http.impl.Http1xServerConnection.handleMessage(Http1xServerConnection.java:145)
	at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:156)
	at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:93)
	at io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtensionHandler.channelRead(WebSocketServerExtensionHandler.java:99)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.vertx.core.http.impl.Http1xUpgradeToH2CHandler.channelRead(Http1xUpgradeToH2CHandler.java:116)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.vertx.core.http.impl.Http1xOrH2CHandler.end(Http1xOrH2CHandler.java:61)
	at io.vertx.core.http.impl.Http1xOrH2CHandler.channelRead(Http1xOrH2CHandler.java:38)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
	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)
```

### How to Reproduce?

[reactive-cache-reproducer.zip](https://github.com/quarkusio/quarkus/files/7720743/reactive-cache-reproducer.zip)

Use attached reproducer:
```
unzip reactive-cache-reproducer.zip
cd reactive-cache-reproducer
mvnw clean test
```

### Output of `uname -a` or `ver`

Linux 5.15.6-200.fc35.x86_64

### Output of `java -version`

openjdk version "11.0.11" 2021-04-20 OpenJDK Runtime Environment AdoptOpenJDK-11.0.11+9 (build 11.0.11+9) OpenJDK 64-Bit Server VM AdoptOpenJDK-11.0.11+9 (build 11.0.11+9, mixed mode)

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

999-SNAPSHOT (2.6 - main)

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

Apache Maven 3.8.1 (05c21c65bdfed0f71a2f2ada8b84da59348c4c5d)

### Additional information

_No response_

Changed files:
extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/UniValueTest.java
extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/CacheResultInterceptor.java

diff --git a/extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/UniValueTest.java b/extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/UniValueTest.java
index 824a65edde1..a75ec898955 100644
--- a/extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/UniValueTest.java
+++ b/extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/UniValueTest.java
@@ -33,17 +33,16 @@ public class UniValueTest {
     public void test() {
         // STEP 1
         // Action: a method annotated with @CacheResult and returning a Uni is called.
-        // Expected effect: the method is invoked and an UnresolvedUniValue is cached.
+        // Expected effect: the method is not invoked, as Uni is lazy
         // Verified by: invocations counter and CacheResultInterceptor log.
         Uni<String> uni1 = cachedService.cachedMethod(KEY);
-        assertEquals(1, cachedService.getInvocations());
+        assertEquals(0, cachedService.getInvocations());
 
         // STEP 2
         // Action: same call as STEP 1.
-        // Expected effect: the method is invoked because the key is associated with a cached UnresolvedUniValue.
-        // Verified by: invocations counter and CacheResultInterceptor log.
+        // Expected effect: the method is not invoked, as Uni is lazy
         Uni<String> uni2 = cachedService.cachedMethod(KEY);
-        assertEquals(2, cachedService.getInvocations());
+        assertEquals(0, cachedService.getInvocations());
 
         // STEP 3
         // Action: the Uni returned in STEP 1 is subscribed to and we wait for an item event to be fired.
@@ -51,22 +50,25 @@ public void test() {
         // Verified by: subscriptions counter and CaffeineCache log.
         String emittedItem1 = uni1.await().indefinitely();
         assertEquals("1", emittedItem1); // This checks the subscriptions counter value.
+        //the method would be called to resolve the value
+        assertEquals(1, cachedService.getInvocations());
 
         // STEP 4
         // Action: the Uni returned in STEP 2 is subscribed to and we wait for an item event to be fired.
         // Expected effect: the emitted item from STEP 3 is replaced with the emitted item from this step in the cache.
         // Verified by: subscriptions counter, CaffeineCache log and different objects references between STEPS 3 and 4 emitted items.
         String emittedItem2 = uni2.await().indefinitely();
-        assertTrue(emittedItem1 != emittedItem2);
-        assertEquals("2", emittedItem2); // This checks the subscriptions counter value.
+        assertTrue(emittedItem1 == emittedItem2);
+        assertEquals("1", emittedItem2); // This checks the subscriptions counter value.
+        assertEquals(1, cachedService.getInvocations());
 
         // STEP 5
         // Action: same call as STEP 2 but we immediately subscribe to the returned Uni and wait for an item event to be fired.
         // Expected effect: the method is not invoked and the emitted item cached during STEP 4 is returned.
         // Verified by: invocations and subscriptions counters, same object reference between STEPS 4 and 5 emitted items.
         String emittedItem3 = cachedService.cachedMethod(KEY).await().indefinitely();
-        assertEquals(2, cachedService.getInvocations());
-        assertEquals("2", emittedItem3); // This checks the subscriptions counter value.
+        assertEquals(1, cachedService.getInvocations());
+        assertEquals("1", emittedItem3); // This checks the subscriptions counter value.
         assertTrue(emittedItem2 == emittedItem3);
 
         // STEP 6
@@ -74,16 +76,16 @@ public void test() {
         // Expected effect: the method is invoked and an UnresolvedUniValue is cached.
         // Verified by: invocations and subscriptions counters, CacheResultInterceptor log and different objects references between STEPS 5 and 6 emitted items.
         String emittedItem4 = cachedService.cachedMethod("another-key").await().indefinitely();
-        assertEquals(3, cachedService.getInvocations());
-        assertEquals("3", emittedItem4); // This checks the subscriptions counter value.
+        assertEquals(2, cachedService.getInvocations());
+        assertEquals("2", emittedItem4); // This checks the subscriptions counter value.
         assertTrue(emittedItem3 != emittedItem4);
     }
 
     @ApplicationScoped
     static class CachedService {
 
-        private int invocations;
-        private int subscriptions;
+        private volatile int invocations;
+        private volatile int subscriptions;
 
         @CacheResult(cacheName = "test-cache")
         public Uni<String> cachedMethod(String key) {
diff --git a/extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/CacheResultInterceptor.java b/extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/CacheResultInterceptor.java
index 034dc77df89..e780bbda786 100644
--- a/extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/CacheResultInterceptor.java
+++ b/extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/CacheResultInterceptor.java
@@ -42,43 +42,71 @@ public Object intercept(InvocationContext invocationContext) throws Throwable {
         }
 
         try {
-
-            Uni<Object> cacheValue = cache.get(key, new Function<Object, Object>() {
-                @Override
-                public Object apply(Object k) {
-                    try {
-                        if (Uni.class.isAssignableFrom(invocationContext.getMethod().getReturnType())) {
-                            LOGGER.debugf("Adding %s entry with key [%s] into cache [%s]",
-                                    UnresolvedUniValue.class.getSimpleName(), key, binding.cacheName());
-                            return UnresolvedUniValue.INSTANCE;
-                        } else {
-                            return invocationContext.proceed();
+            final boolean isUni = Uni.class.isAssignableFrom(invocationContext.getMethod().getReturnType());
+            if (isUni) {
+                Uni<Object> ret = cache.get(key, new Function<Object, Object>() {
+                    @Override
+                    public Object apply(Object k) {
+                        LOGGER.debugf("Adding %s entry with key [%s] into cache [%s]",
+                                UnresolvedUniValue.class.getSimpleName(), key, binding.cacheName());
+                        return UnresolvedUniValue.INSTANCE;
+                    }
+                }).onItem().transformToUni(o -> {
+                    if (o == UnresolvedUniValue.INSTANCE) {
+                        try {
+                            return ((Uni<Object>) invocationContext.proceed())
+                                    .onItem().call(emittedValue -> cache.replaceUniValue(key, emittedValue));
+                        } catch (CacheException e) {
+                            throw e;
+                        } catch (Exception e) {
+                            throw new CacheException(e);
                         }
-                    } catch (Throwable e) {
-                        throw new CacheException(e);
+                    } else {
+                        return Uni.createFrom().item(o);
                     }
+                });
+                if (binding.lockTimeout() <= 0) {
+                    return ret;
                 }
-            });
+                return ret.ifNoItem().after(Duration.ofMillis(binding.lockTimeout())).recoverWithUni(() -> {
+                    try {
+                        return (Uni<?>) invocationContext.proceed();
+                    } catch (CacheException e) {
+                        throw e;
+                    } catch (Exception e) {
+                        throw new CacheException(e);
+                    }
+                });
 
-            Object value;
-            if (binding.lockTimeout() <= 0) {
-                value = cacheValue.await().indefinitely();
             } else {
-                try {
-                    /*
-                     * If the current thread started the cache value computation, then the computation is already finished since
-                     * it was done synchronously and the following call will never time out.
-                     */
-                    value = cacheValue.await().atMost(Duration.ofMillis(binding.lockTimeout()));
-                } catch (TimeoutException e) {
-                    // TODO: Add statistics here to monitor the timeout.
-                    return invocationContext.proceed();
+                Uni<Object> cacheValue = cache.get(key, new Function<Object, Object>() {
+                    @Override
+                    public Object apply(Object k) {
+                        try {
+                            return invocationContext.proceed();
+                        } catch (CacheException e) {
+                            throw e;
+                        } catch (Throwable e) {
+                            throw new CacheException(e);
+                        }
+                    }
+                });
+                Object value;
+                if (binding.lockTimeout() <= 0) {
+                    value = cacheValue.await().indefinitely();
+                } else {
+                    try {
+                        /*
+                         * If the current thread started the cache value computation, then the computation is already finished
+                         * since
+                         * it was done synchronously and the following call will never time out.
+                         */
+                        value = cacheValue.await().atMost(Duration.ofMillis(binding.lockTimeout()));
+                    } catch (TimeoutException e) {
+                        // TODO: Add statistics here to monitor the timeout.
+                        return invocationContext.proceed();
+                    }
                 }
-            }
-
-            if (Uni.class.isAssignableFrom(invocationContext.getMethod().getReturnType())) {
-                return resolveUni(invocationContext, cache, key, value);
-            } else {
                 return value;
             }
 
@@ -90,14 +118,4 @@ public Object apply(Object k) {
             }
         }
     }
-
-    private Object resolveUni(InvocationContext invocationContext, AbstractCache cache, Object key, Object value)
-            throws Exception {
-        if (value == UnresolvedUniValue.INSTANCE) {
-            return ((Uni<Object>) invocationContext.proceed())
-                    .onItem().call(emittedValue -> cache.replaceUniValue(key, emittedValue));
-        } else {
-            return Uni.createFrom().item(value);
-        }
-    }
 }