Reproducer for leaking transactions in XA mode

https://issues.redhat.com/browse/QUARKUS-4185

sql-db/narayana-transactions/src/main/java/io/quarkus/ts/service/ServiceResource.java

@@ -1,17 +1,23 @@
 package io.quarkus.ts.service;
 
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 
 import jakarta.inject.Inject;
+import jakarta.ws.rs.GET;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.core.Response;
 
+import org.jboss.logging.Logger;
+
 import io.agroal.api.AgroalDataSource;
 
 @Path("/service")
 public class ServiceResource {
+    private static final Logger LOG = Logger.getLogger(ServiceResource.class);
+
     @Inject
     AgroalDataSource defaultDataSource;
 
@@ -25,4 +31,20 @@ public Response grant(String user) throws SQLException {
             return Response.status(status).build();
         }
     }
+
+    @Path("/connections")
+    @GET
+    public Response connections() {
+        try (Statement statement = defaultDataSource.getConnection().createStatement()) {
+            ResultSet set = statement.executeQuery("SELECT COUNT(*) from sys.dm_exec_connections;");
+            if (set.next()) {
+                int count = set.getInt(1);
+                return Response.ok(count).build();
+            }
+            return Response.serverError().build();
+        } catch (Exception e) {
+            LOG.error("Failed to retrieve number of connections", e);
+            return Response.serverError().build();
+        }
+    }
 }

sql-db/narayana-transactions/src/test/java/io/quarkus/ts/transactions/MssqlTransactionGeneralUsageIT.java

@@ -3,20 +3,30 @@
 import static io.quarkus.test.services.containers.DockerContainerManagedResource.DOCKER_INNER_CONTAINER;
 
 import java.io.IOException;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 
+import org.apache.http.HttpStatus;
+import org.jboss.logging.Logger;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
 
 import io.quarkus.test.bootstrap.RestService;
 import io.quarkus.test.bootstrap.SqlServerService;
 import io.quarkus.test.scenarios.QuarkusScenario;
 import io.quarkus.test.scenarios.annotations.DisabledOnFipsAndJava17;
+import io.quarkus.test.scenarios.annotations.DisabledOnQuarkusSnapshot;
 import io.quarkus.test.services.QuarkusApplication;
 import io.quarkus.test.services.SqlServerContainer;
 import io.quarkus.ts.transactions.recovery.TransactionExecutor;
+import io.restassured.response.Response;
 
 @DisabledOnFipsAndJava17(reason = "https://github.com/quarkusio/quarkus/issues/40813")
 @QuarkusScenario
 public class MssqlTransactionGeneralUsageIT extends TransactionCommons {
+    private static final Logger LOG = Logger.getLogger(MssqlTransactionGeneralUsageIT.class);
 
     @SqlServerContainer
     static SqlServerService database = new SqlServerService().onPostStart(service -> {
@@ -65,4 +75,24 @@ protected Operation[] getExpectedJdbcOperations() {
                 new Operation(actualOperationName -> actualOperationName.startsWith("UPDATE msdb.")) };
     }
 
+    @Test
+    @DisabledOnQuarkusSnapshot(reason = "The test is too long to run it on CI")
+    @Tag("QUARKUS-4185")
+    //on average, there is one leaking connection every 2 minutes
+    void connectionLeak() throws InterruptedException {
+        int before = getConnections();
+        Duration later = Duration.of(4L, ChronoUnit.MINUTES).plus(Duration.ofSeconds(15));
+        LOG.warn("Waiting for " + later.toSeconds() + " seconds to check for leaking connections");
+        Thread.sleep(later.toMillis());
+        int after = getConnections();
+        Assertions.assertFalse(after - before > 1, //single additional connection may be open temporary
+                "Connections are leaking, was: " + before + " now: " + after);
+
+    }
+
+    private int getConnections() {
+        Response response = getApp().given().get("/service/connections");
+        Assertions.assertEquals(HttpStatus.SC_OK, response.statusCode());
+        return Integer.parseInt(response.asString());
+    }
 }