Skip to content

Fix afterAcquire hook failure leaking connection lease and metrics #1700

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tsegismont merged 1 commit into from
Sep 17, 2026
+142 −80
Merged

Fix afterAcquire hook failure leaking connection lease and metrics #1700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,29 +270,35 @@ class PoolRequest implements PoolWaiter.Listener<PooledConnection>, Completable<

@Override
public void complete(Lease<PooledConnection> lease, Throwable failure) {
if (timerID != -1L && !vertx.cancelTimer(timerID)) {
lease.recycle();
} else {
if (failure == null) {
if (failure == null) {
if (timerID != -1L && !vertx.cancelTimer(timerID)) {
lease.recycle();
} else {
if (afterAcquire != null) {
afterAcquire.apply(lease.get().conn).onComplete(ar2 -> {
if (ar2.succeeded()) {
handle(lease);
} else {
// Should we do some cleanup ?
handler.fail(failure);
fail(lease, ar2.cause());
}
});
} else {
handle(lease);
}
} else {
dequeueMetric(metric);
handler.fail(failure);
}
} else {
fail(null, failure);
}
}

private void fail(Lease<PooledConnection> lease, Throwable cause) {
if (lease != null) {
lease.recycle();
}
dequeueMetric(metric);
handler.fail(cause);
}

private void handle(Lease<PooledConnection> lease) {
dequeueMetric(metric);
PooledConnection pooled = lease.get();
Expand Down
198 changes: 127 additions & 71 deletions vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/spi/backend/DriverBaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import io.vertx.sqlclient.RowIterator;
import io.vertx.sqlclient.RowSet;
import io.vertx.sqlclient.SqlConnectOptions;
import io.vertx.sqlclient.SqlConnection;
import io.vertx.sqlclient.desc.ColumnDescriptor;
import io.vertx.sqlclient.impl.RowBase;
import io.vertx.sqlclient.spi.connection.Connection;
Expand All @@ -28,6 +29,8 @@
import org.junit.Test;

import java.sql.JDBCType;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.stream.Collector;

Expand Down Expand Up @@ -61,103 +64,115 @@ public JDBCType jdbcType() {
}
}

@Test
public void testSimple() {
private static Connection fakeConnection() {
return new Connection() {
@Override
public TracingPolicy tracingPolicy() {
return null;
}
@Override
public SocketAddress server() {
return null;
}
@Override
public String database() {
return "";
}
@Override
public String user() {
return "";
}
@Override
public ClientMetrics metrics() {
return null;
}
@Override
public void init(ConnectionContext context) {
}
@Override
public boolean isSsl() {
return false;
}
@Override
public boolean isValid() {
return true;
}
@Override
public int pipeliningLimit() {
return 1;
}
@Override
public DatabaseMetadata databaseMetadata() {
throw new UnsupportedOperationException();
}
@Override
public void close(ConnectionContext holder, Completable<Void> promise) {
promise.succeed();
}
@Override
public <R> void schedule(CommandBase<R> cmd, Completable<R> handler) {
if (cmd instanceof SimpleQueryCommand) {
SimpleQueryCommand simpleQueryCmd = (SimpleQueryCommand) cmd;
scheduleQueryCommand(simpleQueryCmd, (Completable) handler);
} else {
handler.fail(new UnsupportedOperationException());
}
}
private <T, A> void scheduleQueryCommand(SimpleQueryCommand<T> simpleQuery, Completable<Boolean> handler) {
QueryResultHandler<T> qrh = simpleQuery.resultHandler();
Collector<Row, A, T> collector = (Collector<Row, A, T>) simpleQuery.collector();
A container = collector.supplier().get();
BiConsumer<A, Row> accumulator = collector.accumulator();
RowDescriptorBase rowDescriptor = new RowDescriptorBase(new ColumnDescriptor[]{new VarcharColumnDescriptor("value")});
Row row = new RowBase(rowDescriptor);
row.addValue("Hello " + simpleQuery.sql());
accumulator.accept(container, row);
T result = collector.finisher().apply(container);
qrh.handleResult(0, 1, rowDescriptor
, result, null);
handler.succeed(true);
}
};
}

DriverBase<SqlConnectOptions> driver = new DriverBase<SqlConnectOptions>("generic") {
private static DriverBase<SqlConnectOptions> createDriver(
java.util.function.Function<Connection, Future<Void>> afterAcquire,
java.util.function.Function<Connection, Future<Void>> beforeRecycle) {
return new DriverBase<>("generic", afterAcquire, beforeRecycle) {
@Override
public ConnectionFactory<SqlConnectOptions> createConnectionFactory(Vertx vertx, NetClientOptions transportOptions) {
return new ConnectionFactory<>() {
@Override
public Future<Connection> connect(Context context, SqlConnectOptions options) {
return Future.succeededFuture(new Connection() {
@Override
public TracingPolicy tracingPolicy() {
return null;
}
@Override
public SocketAddress server() {
return null;
}
@Override
public String database() {
return "";
}
@Override
public String user() {
return "";
}
@Override
public ClientMetrics metrics() {
return null;
}
@Override
public void init(ConnectionContext context) {
}
@Override
public boolean isSsl() {
return false;
}
@Override
public boolean isValid() {
return true;
}
@Override
public int pipeliningLimit() {
return 1;
}
@Override
public DatabaseMetadata databaseMetadata() {
throw new UnsupportedOperationException();
}
@Override
public void close(ConnectionContext holder, Completable<Void> promise) {
promise.succeed();
}
@Override
public <R> void schedule(CommandBase<R> cmd, Completable<R> handler) {
if (cmd instanceof SimpleQueryCommand) {
SimpleQueryCommand simpleQueryCmd = (SimpleQueryCommand) cmd;
scheduleQueryCommand(simpleQueryCmd, (Completable) handler);
} else {
handler.fail(new UnsupportedOperationException());
}
}
private <T, A> void scheduleQueryCommand(SimpleQueryCommand<T> simpleQuery, Completable<Boolean> handler) {
QueryResultHandler<T> qrh = simpleQuery.resultHandler();
Collector<Row, A, T> collector = (Collector<Row, A, T>) simpleQuery.collector();
A container = collector.supplier().get();
BiConsumer<A, Row> accumulator = collector.accumulator();
RowDescriptorBase rowDescriptor = new RowDescriptorBase(new ColumnDescriptor[]{new VarcharColumnDescriptor("value")});
Row row = new RowBase(rowDescriptor);
row.addValue("Hello " + simpleQuery.sql());
accumulator.accept(container, row);
T result = collector.finisher().apply(container);
qrh.handleResult(0, 1, rowDescriptor
, result, null);
handler.succeed(true);
}
});
return Future.succeededFuture(fakeConnection());
}
@Override
public void close(Completable<Void> completable) {
completable.succeed();
}
};
}

@Override
public SqlConnectOptions parseConnectionUri(String uri) {
throw new UnsupportedOperationException();
}

@Override
public boolean acceptsOptions(SqlConnectOptions connectOptions) {
return true;
}

@Override
public SqlConnectOptions downcast(SqlConnectOptions connectOptions) {
return connectOptions;
}
};
}

@Test
public void testSimple() {
DriverBase<SqlConnectOptions> driver = createDriver(null, null);

Vertx vertx = Vertx.vertx();

Expand All @@ -177,4 +192,45 @@ public SqlConnectOptions downcast(SqlConnectOptions connectOptions) {
vertx.close().await();
}
}

@Test
public void testAfterAcquireFailureReleasesConnection() {
AtomicBoolean shouldFail = new AtomicBoolean(true);
AtomicInteger beforeRecycleCount = new AtomicInteger();
RuntimeException hookError = new RuntimeException("afterAcquire failed");

DriverBase<SqlConnectOptions> driver = createDriver(
conn -> {
if (shouldFail.get()) {
return Future.failedFuture(hookError);
}
return Future.succeededFuture();
},
conn -> {
beforeRecycleCount.incrementAndGet();
return Future.succeededFuture();
});

Vertx vertx = Vertx.vertx();

try {
Pool pool = driver.createPool(vertx, () -> Future.succeededFuture(new SqlConnectOptions()),
new PoolOptions().setMaxSize(1), new NetClientOptions(), null);

try {
pool.getConnection().await();
fail("Should have failed");
} catch (Exception e) {
assertEquals(hookError, e);
}

shouldFail.set(false);
SqlConnection conn = pool.getConnection().await();
assertNotNull(conn);
conn.close().await();
assertEquals(1, beforeRecycleCount.get());
} finally {
vertx.close().await();
}
}
}