fix: replace syncronization in Connection.close with compareAndSet

Statement can be synchronized during query execution, and we don't want
.close() to wait for the statement completion.

pgjdbc/src/main/java/org/postgresql/jdbc/PgStatement.java

@@ -520,7 +520,10 @@ public void setCursorName(String name) throws SQLException {
     // No-op.
   }
 
-  private volatile boolean isClosed = false;
+  private volatile int isClosed = 0;
+  private static final AtomicIntegerFieldUpdater<PgStatement> IS_CLOSED_UPDATER =
+      AtomicIntegerFieldUpdater.newUpdater(
+          PgStatement.class, "isClosed");
 
   @Override
   public int getUpdateCount() throws SQLException {
@@ -673,11 +676,8 @@ public void clearWarnings() throws SQLException {
    */
   public final void close() throws SQLException {
     // closing an already closed Statement is a no-op.
-    synchronized (this) {
-      if (isClosed) {
-        return;
-      }
-      isClosed = true;
+    if (!IS_CLOSED_UPDATER.compareAndSet(this, 0, 1)) {
+      return;
     }
 
     cancel();
@@ -1138,7 +1138,7 @@ public long executeLargeUpdate(String sql, String @Nullable [] columnNames) thro
   }
 
   public boolean isClosed() throws SQLException {
-    return isClosed;
+    return isClosed == 1;
   }
 
   public void setPoolable(boolean poolable) throws SQLException {

pgjdbc/src/test/java/org/postgresql/test/jdbc2/StatementTest.java

@@ -32,7 +32,9 @@
 import java.sql.SQLException;
 import java.sql.SQLWarning;
 import java.sql.Statement;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Random;
@@ -861,7 +863,7 @@ public void testMultipleCancels() throws Exception {
   public void testCancelQueryWithBrokenNetwork() throws SQLException, IOException, InterruptedException {
     // check that stmt.cancel() doesn't hang forever if the network is broken
 
-    ExecutorService executor = Executors.newSingleThreadExecutor();
+    ExecutorService executor = Executors.newCachedThreadPool();
 
     try (StrangeProxyServer proxyServer = new StrangeProxyServer(TestUtil.getServer(), TestUtil.getPort())) {
       Properties props = new Properties();
@@ -875,6 +877,9 @@ public void testCancelQueryWithBrokenNetwork() throws SQLException, IOException,
         proxyServer.stopForwardingAllClients();
 
         stmt.cancel();
+        // Note: network is still inaccessible, so the statement execution is still in progress.
+        // So we abort the connection to allow implicit conn.close()
+        conn.abort(executor);
       }
     }
 
@@ -940,12 +945,13 @@ public Void call() throws Exception {
   }
 
   @Test(timeout = 10000)
-  public void testConcurrentIsValid() throws InterruptedException {
+  public void testConcurrentIsValid() throws Throwable {
     ExecutorService executor = Executors.newCachedThreadPool();
     try {
+      List<Future<?>> results = new ArrayList<>();
       Random rnd = new Random();
       for (int i = 0; i < 10; i++) {
-        executor.submit(() -> {
+        Future<?> future = executor.submit(() -> {
           try {
             for (int j = 0; j < 50; j++) {
               con.isValid(1);
@@ -969,7 +975,14 @@ public void testConcurrentIsValid() throws InterruptedException {
             throw new RuntimeException(e);
           }
         });
+        results.add(future);
+      }
+      for (Future<?> result : results) {
+        // Propagate exception if any
+        result.get();
       }
+    } catch (ExecutionException e) {
+      throw e.getCause();
     } finally {
       executor.shutdown();
       executor.awaitTermination(10, TimeUnit.SECONDS);