Repository: quarkusio/quarkus
Issue: Hibernate Reactive: ReactiveSessionProducer blocks event loop thread

## Describe the bug
Most likely since Quarkus 2.0.0.Alpha2 with the introduction of [these changes](https://github.com/quarkusio/quarkus/pull/16923), Hibernate Reactive can sometimes block the event loop thread forever if injecting `Mutiny.Session` via the `ReactiveSessionProducer`, making the entire application unresponsive.

### Expected behavior
Hibernate Reactive should properly dispose its sessions without blocking.

### Actual behavior
It is actually blocking the event loop thread. See the following thread dump from one of our applications. The thread remains in this state forever. As a consequence, the entire application cannot handle any incoming requests anymore.

```
"vert.x-eventloop-thread-1" #25 prio=5 os_prio=0 cpu=3312.11ms elapsed=815.13s tid=0x00007f778f573000 nid=0x2f waiting on condition  [0x00007f772698d000]
   java.lang.Thread.State: WAITING (parking)
    at jdk.internal.misc.Unsafe.park(java.base@11.0.9.1/Native Method)
    - parking to wait for  <0x00000000ffadedf8> (a java.util.concurrent.CompletableFuture$Signaller)
    at java.util.concurrent.locks.LockSupport.park(java.base@11.0.9.1/LockSupport.java:194)
    at java.util.concurrent.CompletableFuture$Signaller.block(java.base@11.0.9.1/CompletableFuture.java:1796)
    at java.util.concurrent.ForkJoinPool.managedBlock(java.base@11.0.9.1/ForkJoinPool.java:3128)
    at java.util.concurrent.CompletableFuture.waitingGet(java.base@11.0.9.1/CompletableFuture.java:1823)
    at java.util.concurrent.CompletableFuture.join(java.base@11.0.9.1/CompletableFuture.java:2043)
    at io.smallrye.context.CompletableFutureWrapper.join(CompletableFutureWrapper.java:166)
    at io.quarkus.hibernate.reactive.runtime.ReactiveSessionProducer.disposeMutinySession(ReactiveSessionProducer.java:54)
    at io.quarkus.hibernate.reactive.runtime.ReactiveSessionProducer_ProducerMethod_createMutinySession_1321d110ee9e92bda147899150401e0a136779c7_Bean.destroy(ReactiveSessionProducer_ProducerMethod_createMutinySession_1321d110ee9e92bda147899150401e0a136779c7_Bean.zig:180)
    at io.quarkus.hibernate.reactive.runtime.ReactiveSessionProducer_ProducerMethod_createMutinySession_1321d110ee9e92bda147899150401e0a136779c7_Bean.destroy(ReactiveSessionProducer_ProducerMethod_createMutinySession_1321d110ee9e92bda147899150401e0a136779c7_Bean.zig:212)
    at io.quarkus.arc.impl.InstanceHandleImpl.destroyInternal(InstanceHandleImpl.java:90)
    at io.quarkus.arc.impl.ContextInstanceHandleImpl.destroy(ContextInstanceHandleImpl.java:20)
    at io.quarkus.arc.impl.RequestContext.destroyContextElement(RequestContext.java:184)
    at io.quarkus.arc.impl.RequestContext$$Lambda$538/0x0000000100730c40.accept(Unknown Source)
    at java.util.concurrent.ConcurrentHashMap.forEach(java.base@11.0.9.1/ConcurrentHashMap.java:1603)
    at io.quarkus.arc.impl.RequestContext.destroy(RequestContext.java:170)
    - locked <0x00000000ffa55538> (a java.util.concurrent.ConcurrentHashMap)
    at io.quarkus.arc.impl.RequestContext.destroy(RequestContext.java:154)
    at io.quarkus.resteasy.reactive.common.runtime.ArcThreadSetupAction$1.close(ArcThreadSetupAction.java:27)
    at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.close(AbstractResteasyReactiveContext.java:101)
    at org.jboss.resteasy.reactive.server.core.ResteasyReactiveRequestContext.close(ResteasyReactiveRequestContext.java:358)
    at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:182)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(java.base@11.0.9.1/Thread.java:829)
```


## To Reproduce

Haven't yet been able to create a reproducer, but most likely happens when some failure happens in a reactive chain where HR was previously used (not necessarily in Hibernate itself).

### Configuration
```properties
# Add your application.properties here, if applicable.

```

### Screenshots
(If applicable, add screenshots to help explain your problem.)

## Environment (please complete the following information):

### Output of `uname -a` or `ver`
Microsoft Windows [Version 10.0.18363.1556]

### Output of `java -version`
OpenJDK 64-Bit Server VM Corretto-11.0.10.9.1 (build 11.0.10+9-LTS, mixed mode)

### GraalVM version (if different from Java)
N/A

### Quarkus version or git rev
2.0.0.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)
Apache Maven 3.6.3

## Additional context
(Add any other context about the problem here.)


Changed files:
extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/ReactiveSessionProducer.java

diff --git a/extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/ReactiveSessionProducer.java b/extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/ReactiveSessionProducer.java
index ead36fc3330..804aa5ae6e8 100644
--- a/extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/ReactiveSessionProducer.java
+++ b/extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/ReactiveSessionProducer.java
@@ -1,5 +1,7 @@
 package io.quarkus.hibernate.reactive.runtime;
 
+import java.util.concurrent.CompletableFuture;
+
 import javax.enterprise.context.RequestScoped;
 import javax.enterprise.inject.Disposes;
 import javax.enterprise.inject.Produces;
@@ -23,13 +25,23 @@ public Mutiny.Session createMutinySession() {
 
     public void disposeMutinySession(@Disposes Mutiny.Session reactiveSession) {
         if (reactiveSession != null) {
-            //We're ignoring the returned CompletionStage!
-            //This could certainly done better but it should be effective enough to
-            //eventually close the session: the connection pool will order the operations
-            //as it won't be able to return connections to other consumers when it's saturated,
-            //ensuring close operations are handled.
-            //N.B. for the Mutiny API we need to make sure to subscribe
-            reactiveSession.close().subscribe().asCompletionStage().join();
+            //N.B. make sure to subscribe as this is a Mutiny based Session:
+            // operations don't happen at all if there is no subscription.
+            final CompletableFuture<Void> closeOperation = reactiveSession.close()
+                    .subscribe()
+                    .asCompletionStage();
+            if (!io.vertx.core.Context.isOnVertxThread()) {
+                //When invoked from blocking code, behave as expected and block on the operation
+                //so to not starve resources with a deferred close.
+                closeOperation.join();
+            }
+            // [else] no need to block. Worst that can happen is that the opening
+            // of a new Mutiny.Session needs to wait for an available connection,
+            // which implicitly orders it as "downstream" from the previous close
+            // to have actually happened as the connection pool is reactive.
+            // Also, if connections are available there is no real need to wait for
+            // it, so this should be good.
         }
     }
+
 }