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 7eaf11b7a..bfbe5cc81 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 @@ -248,7 +248,12 @@ public Future execute(ContextInternal context, CommandBase cmd, long t }); }, t -> { dequeueAndReject(queueMetric); - return Future.failedFuture(t); + if (timerId != -1 && !vertx.cancelTimer(timerId)) { + // The timer already failed the result + return Future.failedFuture(POOL_QUERY_TIMEOUT_EXCEPTION); + } else { + return Future.failedFuture(t); + } }).onComplete(ar -> { if (ar.succeeded()) { res.complete(ar.result()); diff --git a/vertx-sql-client/src/test/java/io/vertx/sqlclient/impl/pool/SqlConnectionPoolTest.java b/vertx-sql-client/src/test/java/io/vertx/sqlclient/impl/pool/SqlConnectionPoolTest.java index adf59f14a..1b9eec1b0 100644 --- a/vertx-sql-client/src/test/java/io/vertx/sqlclient/impl/pool/SqlConnectionPoolTest.java +++ b/vertx-sql-client/src/test/java/io/vertx/sqlclient/impl/pool/SqlConnectionPoolTest.java @@ -30,6 +30,9 @@ import io.vertx.sqlclient.spi.Driver; import org.junit.Test; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -214,4 +217,88 @@ public void testAfterAcquireFailureReleasesConnection() throws Exception { vertx.close(); } } + + @Test + public void testConnectFailureCancelsConnectionTimeout() throws Exception { + RuntimeException connectError = new RuntimeException("connect failed"); + + VertxInternal vertx = (VertxInternal) Vertx.vertx(); + List uncaught = new CopyOnWriteArrayList<>(); + vertx.exceptionHandler(uncaught::add); + + try { + SqlConnectionPool pool = createPool(vertx, ctx -> Future.failedFuture(connectError)); + + ContextInternal ctx = vertx.getOrCreateContext(); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + ctx.runOnContext(v -> { + pool.execute(ctx, new CommandBase() {}, 1000).onComplete(ar -> { + failure.set(ar.cause()); + latch.countDown(); + }); + }); + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertEquals(connectError, failure.get()); + + // Let the connection timeout elapse + Thread.sleep(1500); + assertEquals(Collections.emptyList(), uncaught); + + pool.close(); + } finally { + vertx.close(); + } + } + + @Test + public void testConnectFailureAfterConnectionTimeout() throws Exception { + RuntimeException connectError = new RuntimeException("connect failed"); + Promise connectPromise = Promise.promise(); + + VertxInternal vertx = (VertxInternal) Vertx.vertx(); + List uncaught = new CopyOnWriteArrayList<>(); + vertx.exceptionHandler(uncaught::add); + + try { + SqlConnectionPool pool = createPool(vertx, ctx -> connectPromise.future()); + + ContextInternal ctx = vertx.getOrCreateContext(); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + ctx.runOnContext(v -> { + pool.execute(ctx, new CommandBase() {}, 100).onComplete(ar -> { + failure.set(ar.cause()); + latch.countDown(); + }); + }); + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertEquals("Timeout waiting for connection", failure.get().getMessage()); + + ctx.runOnContext(v -> connectPromise.fail(connectError)); + Thread.sleep(300); + assertEquals(Collections.emptyList(), uncaught); + + pool.close(); + } finally { + vertx.close(); + } + } + + private static SqlConnectionPool createPool(VertxInternal vertx, java.util.function.Function> connectionProvider) { + return new SqlConnectionPool( + connectionProvider, + () -> null, + null, + null, + null, + vertx, + 0, + 0, + 1, + false, + -1, + 0 + ); + } }