Merge pull request #531 from Bit-Quill/issue-527

fix: statement.cancel() waits for statement to finish

src/main/user-impl/java/com/mysql/cj/jdbc/ha/ConnectionProxy.java

@@ -52,9 +52,12 @@
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.sql.SQLException;
-import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 
 /**
@@ -67,6 +70,7 @@ public class ConnectionProxy implements ICurrentConnectionProvider, InvocationHa
   protected static final Log NULL_LOGGER = new NullLogger(Log.LOGGER_INSTANCE_NAME);
   static final String METHOD_EQUALS = "equals";
   private static final String METHOD_HASH_CODE = "hashCode";
+  private final ReentrantLock lock = new ReentrantLock();
   private final JdbcPropertySetImpl connProps = new JdbcPropertySetImpl();
   /** The logger we're going to use. */
   protected transient Log log = NULL_LOGGER;
@@ -76,7 +80,6 @@ public class ConnectionProxy implements ICurrentConnectionProvider, InvocationHa
   private HostInfo currentHostInfo;
   private JdbcConnection currentConnection;
   private Class<?> currentConnectionClass;
-
   public ConnectionProxy(ConnectionUrl connectionUrl) throws SQLException {
     this(connectionUrl, null);
   }
@@ -314,6 +317,13 @@ class JdbcInterfaceProxy implements InvocationHandler {
     private final Object invokeOn;
     private final Class<?> invokeOnClass;
 
+    private final Set<String> asynchronousMethods =
+        Collections.unmodifiableSet(new HashSet<String>() {
+          {
+            add("cancel");
+          }
+        });
+
     JdbcInterfaceProxy(Object toInvokeOn) {
       this.invokeOn = toInvokeOn;
       this.invokeOnClass = toInvokeOn == null ? null : toInvokeOn.getClass();
@@ -347,14 +357,21 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
         return executeMethodDirectly(methodName, args);
       }
 
-      synchronized(this.invokeOn) {
+      if (!asynchronousMethods.contains(methodName)) {
+        lock.lock();
+      }
+      try {
         Object result =
             ConnectionProxy.this.pluginManager.execute(
                 this.invokeOnClass,
                 methodName,
                 () -> method.invoke(this.invokeOn, args),
                 args);
         return proxyIfReturnTypeIsJdbcInterface(method.getReturnType(), result);
+      } finally {
+        if (lock.isHeldByCurrentThread()) {
+          lock.unlock();
+        }
       }
     }
   }

src/test/java/testsuite/integration/container/AuroraMysqlIntegrationTest.java

@@ -624,4 +624,32 @@ public void test_QueryTimeoutOnReaderClusterConnection() throws Exception {
       }
     }
   }
+
+  @Test
+  public void test_CancelStatementsAreNotBlocked() {
+    try (final Connection conn = connectToInstance(MYSQL_CLUSTER_URL, MYSQL_PORT)) {
+      Statement stmt = conn.createStatement();
+      Thread thread = new Thread(() -> {
+        try {
+          Thread.sleep(1000);
+          stmt.cancel();
+        } catch (SQLException | InterruptedException e) {
+          fail(e);
+        }
+      });
+
+      final long startTime = System.currentTimeMillis();
+      thread.start();
+      stmt.execute("select sleep(100000000)");
+
+      try {
+        thread.join();
+        assert(System.currentTimeMillis() - startTime < 10000);
+      } catch (InterruptedException e) {
+        fail(e);
+      }
+    } catch (Exception e) {
+      fail(e);
+    }
+  }
 }