Some improvements:

- Found a pool test I could modify (but couldn't validate locally yet)
- Changed the code to return the release the connection, which wasn't happening

Author: pablosaavedra-rappi

vertx-pg-client/src/test/java/io/vertx/pgclient/PgPoolTest.java

@@ -22,6 +22,7 @@
 import io.vertx.core.Handler;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.impl.ContextInternal;
+import io.vertx.core.impl.NoStackTraceThrowable;
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.Repeat;
@@ -592,4 +593,45 @@ private void testConnectionClosedInProvider(TestContext ctx, boolean immediately
       }));
     }));
   }
+
+  @Test
+  public void testConnectionTimeoutWhenExecutingDirectly(TestContext ctx) {
+    PgPool pool = createPool(options, new PoolOptions().setConnectionTimeout(2).setMaxSize(2));
+    final Async latch = ctx.async(2);
+    pool.getConnection(ctx.asyncAssertSuccess(conn -> {
+      conn
+        .query("SELECT id, message from immutable")
+        .execute(ctx.asyncAssertSuccess(rows -> {
+          ctx.assertEquals(12, rows.size());
+          latch.countDown();
+        }));
+    }));
+
+    pool.getConnection(ctx.asyncAssertSuccess(conn -> {
+      conn
+        .query("SELECT id, message from immutable")
+        .execute(ctx.asyncAssertSuccess(rows -> {
+          ctx.assertEquals(12, rows.size());
+          latch.countDown();
+        }));
+    }));
+
+    latch.awaitSuccess();
+    final long timerId = vertx.setTimer(10000L, id -> {
+      ctx.fail("Timeout exceeded without completing");
+    });
+    //Used both connections
+    Async async = ctx.async(10);
+    for (int i = 0; i < 10; i++) {
+      pool
+        .query("SELECT id, message from immutable")
+        .execute(ctx.asyncAssertFailure(t -> {
+          ctx.assertTrue(t instanceof NoStackTraceThrowable);
+          ctx.assertEquals("Timeout", t.getMessage());
+          async.countDown();
+        }));
+    }
+
+    async.handler(v -> vertx.cancelTimer(timerId));
+  }
 }

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/PoolImpl.java

@@ -170,8 +170,20 @@ public Future<SqlConnection> getConnection() {
   @Override
   public <R> Future<R> schedule(ContextInternal context, CommandBase<R> cmd) {
     PromiseInternal<SqlConnectionPool.PooledConnection> promise = context.promise();
+    //Acquires the connection honoring the pool's connection timeout
     acquire(context, connectionTimeout, promise);
-    return promise.future().compose(pooled -> pooled.schedule(context, cmd));
+    return promise.future().compose(pooled -> {
+        //We need to 'init' the connection of close will fail.
+        pooled.init(pooled);
+        return pooled.schedule(context, cmd)
+          .eventually(v -> {
+              Promise<Void> p = Promise.promise();
+              pooled.close(pooled, p);
+              return p.future();
+            }
+          );
+      }
+    );
   }
 
   private void acquire(ContextInternal context, long timeout, Handler<AsyncResult<SqlConnectionPool.PooledConnection>> completionHandler) {