vertx_pool_queue_pending grows indefinitely when connection acquisition fails (#1626) (#1627)

See #1625

We must invoke dequeueMetric in all cases, not just in case of timeout.

Signed-off-by: Thomas Segismont <tsegismont@gmail.com>

Author: tsegismont

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

@@ -243,6 +243,9 @@ public <R> void execute(CommandBase<R> cmd, Completable<R> handler, long timeout
         pooled.refresh();
         lease.recycle();
       });
+    }, t -> {
+      dequeueMetric(metric);
+      return Future.failedFuture(t);
     }).onComplete(ar -> {
       if (ar.succeeded()) {
         handler.succeed(ar.result());
@@ -281,6 +284,7 @@ public void complete(Lease<PooledConnection> lease, Throwable failure) {
               handle(lease);
             }
           } else {
+            dequeueMetric(metric);
             handler.fail(failure);
           }
         }

vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/tck/MetricsTestBase.java

@@ -340,4 +340,65 @@ public void requestReset(Object requestMetric) {
       }));
     });
   }
+
+  @Test
+  public void testGetConnectionFailure(TestContext ctx) throws Exception {
+    testConnectionFailure(ctx, true);
+  }
+
+  @Test
+  public void testPooledConnectionFailure(TestContext ctx) throws Exception {
+    testConnectionFailure(ctx, false);
+  }
+
+  private void testConnectionFailure(TestContext ctx, boolean useGetConnection) throws Exception {
+    AtomicInteger queueSize = new AtomicInteger();
+    List<Object> enqueueMetrics = Collections.synchronizedList(new ArrayList<>());
+    List<Object> dequeueMetrics = Collections.synchronizedList(new ArrayList<>());
+    poolMetrics = new PoolMetrics() {
+      @Override
+      public Object enqueue() {
+        Object metric = new Object();
+        enqueueMetrics.add(metric);
+        queueSize.incrementAndGet();
+        return metric;
+      }
+
+      @Override
+      public void dequeue(Object taskMetric) {
+        dequeueMetrics.add(taskMetric);
+        queueSize.decrementAndGet();
+      }
+
+      @Override
+      public Object begin() {
+        throw new IllegalStateException("Shouldn't be invoked");
+      }
+
+      @Override
+      public void end(Object usageMetric) {
+        throw new IllegalStateException("Shouldn't be invoked");
+      }
+    };
+    PoolOptions poolOptions = new PoolOptions().setMaxSize(1).setName("the-pool");
+    SqlConnectOptions connectOptions = connectOptions().setHost("does.not.exist.com");
+    Pool pool = poolBuilder().with(poolOptions).using(vertx).connectingTo(connectOptions).build();
+    int num = 16;
+    List<Future<?>> futures = new ArrayList<>();
+    for (int i = 0; i < num; i++) {
+      Future<RowSet<Row>> future;
+      if (useGetConnection) {
+        future = pool.withConnection(sqlConn -> sqlConn.query("SELECT * FROM immutable WHERE id=1").execute());
+      } else {
+        future = pool.query("SELECT * FROM immutable WHERE id=1").execute();
+      }
+      futures.add(future);
+    }
+    Future.join(futures).otherwiseEmpty().await(20, SECONDS);
+    ctx.assertEquals(0, queueSize.get());
+    ctx.assertEquals(num, enqueueMetrics.size());
+    ctx.assertEquals(enqueueMetrics, dequeueMetrics);
+    ctx.assertEquals("sql", poolType);
+    ctx.assertEquals("the-pool", poolName);
+  }
 }