Do not mark already closed connection as to be reused

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ReusableConnectionFactory.java

@@ -128,7 +128,7 @@ public void close()
         delegate.close();
     }
 
-    private final class CachedConnection
+    final class CachedConnection
             extends ForwardingConnection
     {
         private final String queryId;
@@ -176,7 +176,7 @@ public void close()
             if (dirty) {
                 delegate.close();
             }
-            else {
+            else if (!delegate.isClosed()) {
                 connections.put(queryId, delegate);
             }
         }

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestReusableConnectionFactory.java

@@ -139,6 +139,28 @@ public void testConnectionIsNotReusedWhenSetToReadOnly()
         }
     }
 
+    @Test
+    public void testConnectionIsNotReusedWhenDelegateIsClosed()
+            throws Exception
+    {
+        MockConnectionFactory mockConnectionFactory = new MockConnectionFactory();
+        ReusableConnectionFactory connectionFactory = new ReusableConnectionFactory(mockConnectionFactory, FOREVER, HUGE_SIZE);
+        Connection connection = connectionFactory.openConnection(ALICE);
+
+        Connection delegate = ((ReusableConnectionFactory.CachedConnection) connection).delegate();
+        delegate.close();
+        connection.close();
+
+        Connection secondConnection = connectionFactory.openConnection(ALICE);
+        Connection secondDelegate = ((ReusableConnectionFactory.CachedConnection) secondConnection).delegate();
+
+        assertThat(delegate).isNotEqualTo(secondDelegate);
+        secondConnection.close();
+
+        assertThat(mockConnectionFactory.openedConnections).isEqualTo(2);
+        assertThat(mockConnectionFactory.closedConnections).isEqualTo(1);
+    }
+
     @Test
     public void testConnectionIsNotReusedWhenAutoCommitDisabled()
             throws Exception
@@ -190,6 +212,12 @@ protected Connection delegate()
             throw new UnsupportedOperationException();
         }
 
+        @Override
+        public boolean isClosed()
+        {
+            return closed;
+        }
+
         @Override
         public void setAutoCommit(boolean autoCommit) {}