Repository: quarkusio/quarkus
Issue: Hibernate Reactive: sessions are not closed, connections leak

ATM, we have the following method to produce `Stage.Session` and `Mutiny.Session` tied to the request scope:

```java
package io.quarkus.hibernate.reactive.runtime;

import java.util.concurrent.CompletionStage;

import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.context.RequestScoped;
import javax.enterprise.inject.Disposes;
import javax.enterprise.inject.Produces;
import javax.inject.Inject;

import org.hibernate.reactive.mutiny.Mutiny;
import org.hibernate.reactive.stage.Stage;

import io.quarkus.arc.DefaultBean;
import io.smallrye.mutiny.Uni;

@ApplicationScoped
public class ReactiveSessionProducer {

    @Inject
    private Stage.SessionFactory reactiveSessionFactory;

    @Inject
    private Mutiny.SessionFactory mutinySessionFactory;

    @Produces
    @RequestScoped
    @DefaultBean
    public CompletionStage<Stage.Session> stageSession() {
        return reactiveSessionFactory.openSession();
    }

    @Produces
    @RequestScoped
    @DefaultBean
    public Uni<Mutiny.Session> mutinySession() {
        return mutinySessionFactory.openSession();
    }

    public void disposeStageSession(@Disposes CompletionStage<Stage.Session> reactiveSession) {
        reactiveSession.whenComplete((s, t) -> {
            if (s != null)
                s.close();
        });
    }

    public void disposeMutinySession(@Disposes Uni<Mutiny.Session> reactiveSession) {
        // TODO: @AGG this is not working, need to check w/ Clement to figure out the proper way
        reactiveSession.on().termination((session, ex, cancelled) -> {
            if (session != null)
                session.close();
        });
    }

}
```

Now, the problem is that we're tying a `CompletionStage<Stage.Session>` (and respectively same for Mutiny) to the request, which is a promise of a session.

The `disposeStageSession` cannot work, because `CompletionStage` are immutable, and the value returned by `reactiveSession.whenComplete()` is never used, so this `s.close()` will never be called.

A similar issue exists for `disposeMutinySession` except it's possible that the `Uni<Mutiny.Session>` opens a new session for each call to `subscribe` (unless `cache()` is called on it, I don't know if it is) whereas for `CompletionStage` it's eager and will have a single session by design.

We can't do `whenComplete`/`on().termination` in `stageSession`/`mutinySession` either because that would close the session before we return it to the user.

To be frank, ATM I don't know how to register a way to auto-close what is lazily obtained by a CS/Uni and tied to the request scope.

Changed files:
integration-tests/hibernate-reactive-db2/src/main/java/io/quarkus/it/hibernate/reactive/db2/HibernateReactiveDB2TestEndpoint.java
integration-tests/hibernate-reactive-mysql/src/main/java/io/quarkus/it/hibernate/reactive/mysql/HibernateReactiveMySQLTestEndpoint.java
integration-tests/hibernate-reactive/src/main/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTestEndpoint.java
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 181d97d9d3b..6814c3e5d2a 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
@@ -34,7 +34,7 @@ public CompletionStage<Stage.Session> stageSession() {
     @RequestScoped
     @DefaultBean
     public Uni<Mutiny.Session> mutinySession() {
-        return mutinySessionFactory.openSession();
+        return mutinySessionFactory.openSession().cache();
     }
 
     public void disposeStageSession(@Disposes CompletionStage<Stage.Session> reactiveSession) {
@@ -45,11 +45,7 @@ public void disposeStageSession(@Disposes CompletionStage<Stage.Session> reactiv
     }
 
     public void disposeMutinySession(@Disposes Uni<Mutiny.Session> reactiveSession) {
-        // TODO: @AGG this is not working, need to check w/ Clement to figure out the proper way
-        reactiveSession.on().termination((session, ex, cancelled) -> {
-            if (session != null)
-                session.close();
-        });
+        reactiveSession.subscribe().with(Mutiny.Session::close);
     }
 
 }
diff --git a/integration-tests/hibernate-reactive-db2/src/main/java/io/quarkus/it/hibernate/reactive/db2/HibernateReactiveDB2TestEndpoint.java b/integration-tests/hibernate-reactive-db2/src/main/java/io/quarkus/it/hibernate/reactive/db2/HibernateReactiveDB2TestEndpoint.java
index b71bfb7ae88..ec058f66f25 100644
--- a/integration-tests/hibernate-reactive-db2/src/main/java/io/quarkus/it/hibernate/reactive/db2/HibernateReactiveDB2TestEndpoint.java
+++ b/integration-tests/hibernate-reactive-db2/src/main/java/io/quarkus/it/hibernate/reactive/db2/HibernateReactiveDB2TestEndpoint.java
@@ -132,9 +132,8 @@ public Uni<String> reactiveUpdate() {
     }
 
     private Uni<RowSet<Row>> populateDB() {
-        return db2Pool.getConnection()
-                .flatMap(c -> c.preparedQuery("DELETE FROM Pig").execute().map(junk -> c))
-                .flatMap(c -> c.preparedQuery("INSERT INTO Pig (id, name) VALUES (5, 'Aloi')").execute());
+        return db2Pool.query("DELETE FROM Pig").execute()
+                .flatMap(junk -> db2Pool.preparedQuery("INSERT INTO Pig (id, name) VALUES (5, 'Aloi')").execute());
     }
 
     private Uni<String> selectNameFromId(Integer id) {
diff --git a/integration-tests/hibernate-reactive-mysql/src/main/java/io/quarkus/it/hibernate/reactive/mysql/HibernateReactiveMySQLTestEndpoint.java b/integration-tests/hibernate-reactive-mysql/src/main/java/io/quarkus/it/hibernate/reactive/mysql/HibernateReactiveMySQLTestEndpoint.java
index 4e1695a16e0..54b589b62c4 100644
--- a/integration-tests/hibernate-reactive-mysql/src/main/java/io/quarkus/it/hibernate/reactive/mysql/HibernateReactiveMySQLTestEndpoint.java
+++ b/integration-tests/hibernate-reactive-mysql/src/main/java/io/quarkus/it/hibernate/reactive/mysql/HibernateReactiveMySQLTestEndpoint.java
@@ -132,9 +132,8 @@ public Uni<String> reactiveUpdate() {
     }
 
     private Uni<RowSet<Row>> populateDB() {
-        return mysqlPool.getConnection()
-                .flatMap(c -> c.preparedQuery("DELETE FROM Pig").execute().map(junk -> c))
-                .flatMap(c -> c.preparedQuery("INSERT INTO Pig (id, name) VALUES (5, 'Aloi')").execute());
+        return mysqlPool.query("DELETE FROM Pig").execute()
+                .flatMap(junk -> mysqlPool.preparedQuery("INSERT INTO Pig (id, name) VALUES (5, 'Aloi')").execute());
     }
 
     private Uni<String> selectNameFromId(Integer id) {
diff --git a/integration-tests/hibernate-reactive/src/main/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTestEndpoint.java b/integration-tests/hibernate-reactive/src/main/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTestEndpoint.java
index 147cc835cd3..a39039a542f 100644
--- a/integration-tests/hibernate-reactive/src/main/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTestEndpoint.java
+++ b/integration-tests/hibernate-reactive/src/main/java/io/quarkus/it/hibernate/reactive/postgresql/HibernateReactiveTestEndpoint.java
@@ -132,9 +132,8 @@ public Uni<String> reactiveUpdate() {
     }
 
     private Uni<RowSet<Row>> populateDB() {
-        return pgPool.getConnection()
-                .flatMap(c -> c.preparedQuery("DELETE FROM Pig").execute().map(junk -> c))
-                .flatMap(c -> c.preparedQuery("INSERT INTO Pig (id, name) VALUES (5, 'Aloi')").execute());
+        return pgPool.query("DELETE FROM Pig").execute()
+                .flatMap(junk -> pgPool.preparedQuery("INSERT INTO Pig (id, name) VALUES (5, 'Aloi')").execute());
     }
 
     private Uni<String> selectNameFromId(Integer id) {