diff --git a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PgPoolTestBase.java b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PgPoolTestBase.java index 0b51462fb..513233bdf 100644 --- a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PgPoolTestBase.java +++ b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PgPoolTestBase.java @@ -30,6 +30,7 @@ import org.junit.Before; import org.junit.Test; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; @@ -256,4 +257,20 @@ public void testWithConnection(TestContext ctx) { } } } + + @Test + public void testPoolQueryTimeoutLeavesWaitQueue(TestContext ctx) { + Pool pool = createPool(options, new PoolOptions() + .setMaxSize(1) + .setMaxWaitQueueSize(1) + .setConnectionTimeout(1) + .setConnectionTimeoutUnit(TimeUnit.SECONDS)); + pool.getConnection().onComplete(ctx.asyncAssertSuccess(conn -> { + pool.query("SELECT 1").execute().onComplete(ctx.asyncAssertFailure(timeout -> { + // The timed out query must not keep its place in the wait queue + pool.query("SELECT 1").execute().onComplete(ctx.asyncAssertSuccess()); + vertx.setTimer(100, id -> conn.close()); + })); + })); + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java index 7f82e3b42..e035b70a9 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java @@ -33,6 +33,7 @@ import io.vertx.sqlclient.spi.protocol.QueryCommandBase; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -211,14 +212,38 @@ private void endMetric(Object metric) { public void execute(CommandBase cmd, Completable handler, long timeout) { ContextInternal context = vertx.getOrCreateContext(); Promise> p = context.promise(); + Object metric = enqueueMetric(); + AtomicReference> waiter = new AtomicReference<>(); long timerId; if (timeout > 0) { - timerId = vertx.setTimer(timeout, t -> handler.fail(POOL_QUERY_TIMEOUT_EXCEPTION)); + timerId = vertx.setTimer(timeout, t -> { + PoolWaiter w = waiter.get(); + if (w == null) { + handler.fail(POOL_QUERY_TIMEOUT_EXCEPTION); + return; + } + // Leave the wait queue before failing, the request must no longer count against its size + pool.cancel(w, (removed, err) -> { + if (err == null && removed) { + dequeueMetric(metric); + } + handler.fail(POOL_QUERY_TIMEOUT_EXCEPTION); + }); + }); } else { timerId = -1; } - Object metric = enqueueMetric(); - pool.acquire(context, 0, p); + pool.acquire(context, new PoolWaiter.Listener<>() { + @Override + public void onEnqueue(PoolWaiter w) { + waiter.set(w); + } + + @Override + public void onConnect(PoolWaiter w) { + waiter.set(w); + } + }, 0, p); p.future().compose(lease -> { dequeueMetric(metric); PooledConnection pooled = lease.get();