Fix socket leak and wrong exception propagation on connection error

See #1672, #1674, #1675

When a connection handles a failure, the underlying Netty socket should be closed, avoiding leak of file descriptors and Netty resources.

Additionally, inflight commands shouldn't fail with ClosedConnectionException but with the actual error.

Some portions of this content were created with the assistance of Claude Code.

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

Author: tsegismont

vertx-mssql-client/src/test/java/io/vertx/tests/mssqlclient/MSSQLQueriesTest.java

@@ -40,8 +40,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.*;
 import static org.hamcrest.MatcherAssert.assertThat;
 
 @RunWith(VertxUnitRunner.class)
@@ -206,4 +205,29 @@ public void testQuerySequences(TestContext ctx) {
         }));
     }));
   }
+
+  @Test
+  public void testUnsupportedXmlTypeErrorPropagation(TestContext ctx) {
+    Async async = ctx.async();
+    connection
+      .query("CREATE TABLE #TempXmlTest (id INT, xmlData XML)")
+      .execute()
+      .onComplete(ctx.asyncAssertSuccess(create -> {
+        connection
+          .query("SELECT * FROM #TempXmlTest")
+          .execute()
+          .onComplete(ctx.asyncAssertFailure(t -> {
+            ctx.verify(v -> {
+              assertThat(t, instanceOf(UnsupportedOperationException.class));
+              assertThat(t.getMessage(), containsString("XML"));
+            });
+            connection
+              .query("SELECT 1")
+              .execute()
+              .onComplete(ctx.asyncAssertFailure(t2 -> {
+                connection.close().onComplete(ctx.asyncAssertSuccess(v -> async.complete()));
+              }));
+          }));
+      }));
+  }
 }

vertx-pg-client/src/test/java/io/vertx/tests/pgclient/TLSTest.java

@@ -11,6 +11,7 @@
 
 package io.vertx.tests.pgclient;
 
+import com.ongres.scram.client.ChannelBindingException;
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.buffer.Buffer;
@@ -25,7 +26,6 @@
 import io.vertx.pgclient.PgConnection;
 import io.vertx.pgclient.SslMode;
 import io.vertx.pgclient.SslNegotiation;
-import io.vertx.sqlclient.ClosedConnectionException;
 import io.vertx.sqlclient.Tuple;
 import io.vertx.tests.pgclient.junit.ContainerPgRule;
 import org.junit.*;
@@ -287,8 +287,7 @@ public void testChannelBindingRequireWithoutSsl(TestContext ctx) {
       .setSslOptions(new ClientSSLOptions().setTrustAll(true));
 
     PgConnection.connect(vertx, options).onComplete(ctx.asyncAssertFailure(err -> {
-      // ctx.assertEquals("Channel bindins is required", err.getMessage());
-      ctx.assertTrue(err instanceof ClosedConnectionException); // TODO: handle ChannelBindingException
+      ctx.assertTrue(err instanceof ChannelBindingException);
       async.complete();
     }));
   }

vertx-pg-client/src/test/java/module-info.java

@@ -2,6 +2,7 @@
 
   requires io.netty.buffer;
   requires io.netty.transport;
+  requires com.ongres.scram.client;
   requires io.vertx.core;
   requires io.vertx.sql.client;
   requires io.vertx.sql.client.pg;

vertx-sql-client-codec/src/main/java/io/vertx/sqlclient/codec/SocketConnectionBase.java

@@ -421,6 +421,8 @@ protected void handleException(Throwable t) {
       DecoderException err = (DecoderException) t;
       t = err.getCause();
     }
+    socket.closeHandler(null);
+    socket.close();
     handleClose(t);
   }
 
@@ -441,11 +443,12 @@ protected void handleClose(Throwable t) {
       if (t != null) {
         reportException(t);
       }
+      Throwable inflightCause = t != null ? t : ClosedConnectionException.INSTANCE;
       CommandMessage<?, ?> msg;
       while ((msg = inflights.poll()) != null) {
-        fail(msg, ClosedConnectionException.INSTANCE);
+        fail(msg, inflightCause);
       }
-      Throwable cause = t == null ? VertxException.noStackTrace(PENDING_CMD_CONNECTION_CORRUPT_MSG) : new VertxException(PENDING_CMD_CONNECTION_CORRUPT_MSG, t);
+      Throwable cause = t == null ? VertxException.noStackTrace(PENDING_CMD_CONNECTION_CORRUPT_MSG) : VertxException.noStackTrace(PENDING_CMD_CONNECTION_CORRUPT_MSG, t);
       CommandBase<?> cmd;
       while ((cmd = pending.poll()) != null) {
         CommandBase c = cmd;