feat(jdbc-mysql): clean up abandoned connection when quarkus app runs…

… in native mode

Make sure that abandoned threads are cleaned up even in native mode using the same algorithm used in
JVM mode.

extensions/jdbc/jdbc-mysql/deployment/src/main/java/io/quarkus/jdbc/mysql/deployment/JDBCMySQLProcessor.java

@@ -7,12 +7,15 @@
 import io.quarkus.deployment.Capabilities;
 import io.quarkus.deployment.annotations.BuildProducer;
 import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
 import io.quarkus.deployment.builditem.NativeImageEnableAllCharsetsBuildItem;
 import io.quarkus.deployment.builditem.NativeImageEnableAllTimeZonesBuildItem;
 import io.quarkus.deployment.builditem.SslNativeConfigBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem;
 import io.quarkus.jdbc.mysql.runtime.MySQLAgroalConnectionConfigurer;
+import io.quarkus.jdbc.mysql.runtime.MySQLRecorder;
 
 public class JDBCMySQLProcessor {
 
@@ -39,6 +42,12 @@ void configureAgroalConnection(BuildProducer<AdditionalBeanBuildItem> additional
         }
     }
 
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    void abandonedConnectionCleanUp(MySQLRecorder recorder) {
+        recorder.startAbandonedConnectionCleanup();
+    }
+
     @BuildStep
     NativeImageResourceBuildItem resource() {
         return new NativeImageResourceBuildItem("com/mysql/cj/util/TimeZoneMapping.properties");

extensions/jdbc/jdbc-mysql/runtime/src/main/java/io/quarkus/jdbc/mysql/runtime/MySQLRecorder.java

@@ -0,0 +1,16 @@
+package io.quarkus.jdbc.mysql.runtime;
+
+import org.graalvm.nativeimage.ImageInfo;
+
+import io.quarkus.jdbc.mysql.runtime.graal.AbandonedConnectionCleanupThread_disable;
+import io.quarkus.runtime.annotations.Recorder;
+
+@Recorder
+public class MySQLRecorder {
+
+    public void startAbandonedConnectionCleanup() {
+        if (ImageInfo.inImageRuntimeCode()) {
+            AbandonedConnectionCleanupThread_disable.startCleanUp();
+        }
+    }
+}

extensions/jdbc/jdbc-mysql/runtime/src/main/java/io/quarkus/jdbc/mysql/runtime/graal/AbandonedConnectionCleanupThread_disable.java

@@ -1,5 +1,13 @@
 package io.quarkus.jdbc.mysql.runtime.graal;
 
+import java.lang.ref.PhantomReference;
+import java.lang.ref.Reference;
+import java.lang.ref.ReferenceQueue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import com.mysql.cj.MysqlConnection;
 import com.mysql.cj.jdbc.AbandonedConnectionCleanupThread;
 import com.mysql.cj.protocol.NetworkResources;
@@ -10,17 +18,97 @@
  * This is to disable connection clean up thread {@link AbandonedConnectionCleanupThread} which launches a thread on a static
  * block.
  * GraalVM is not happy about that. The issue might have been fixed with https://github.com/oracle/graal/pull/1542 but we have
- * to wait for a proper GraalVM release, so we substitute the whole class for now.
- *
- * There is a PR @see <a href="https://github.com/mysql/mysql-connector-j/pull/41">on the mysql repo</a> to enable
- * disabling of this class completely by using a system property. Let's see if we can get rid of this once the PR is merged.
+ * to wait for a proper GraalVM release, so we substitute the class and start the clean up thread manually when running in
+ * native image.
  */
 @Substitute
 @TargetClass(AbandonedConnectionCleanupThread.class)
-final public class AbandonedConnectionCleanupThread_disable {
+final public class AbandonedConnectionCleanupThread_disable implements Runnable {
+
+    private static Set<ConnectionFinalizerPhantomReference> connectionFinalizerPhantomReferences;
+    private static ReferenceQueue<MysqlConnection> mysqlConnectionReferenceQueue;
+
+    private static ExecutorService executorService;
+
+    @Substitute
+    private AbandonedConnectionCleanupThread_disable() {
+    }
+
+    public static void startCleanUp() {
+        connectionFinalizerPhantomReferences = ConcurrentHashMap.newKeySet();
+        mysqlConnectionReferenceQueue = new ReferenceQueue<>();
+        executorService = Executors.newSingleThreadScheduledExecutor();
+        executorService.execute(new AbandonedConnectionCleanupThread_disable());
+    }
+
+    @Substitute
+    public void run() {
+        for (;;) {
+            try {
+                Reference<? extends MysqlConnection> reference = mysqlConnectionReferenceQueue.remove(5000);
+                if (reference != null) {
+                    finalizeResourceAndRemoveReference((ConnectionFinalizerPhantomReference) reference);
+                }
+            } catch (InterruptedException e) {
+                synchronized (mysqlConnectionReferenceQueue) {
+                    Reference<? extends MysqlConnection> reference;
+                    while ((reference = mysqlConnectionReferenceQueue.poll()) != null) {
+                        finalizeResourceAndRemoveReference((ConnectionFinalizerPhantomReference) reference);
+                    }
+                    connectionFinalizerPhantomReferences.clear();
+                }
+
+                return;
+            } catch (Exception ex) {
+            }
+        }
+    }
+
+    @Substitute
+    public static void checkedShutdown() {
+        executorService.shutdownNow();
+    }
+
+    @Substitute
+    public static void uncheckedShutdown() {
+        executorService.shutdownNow();
+    }
 
     @Substitute
     protected static void trackConnection(MysqlConnection conn, NetworkResources io) {
-        // do nothing
+        synchronized (mysqlConnectionReferenceQueue) {
+            ConnectionFinalizerPhantomReference reference = new ConnectionFinalizerPhantomReference(conn, io,
+                    mysqlConnectionReferenceQueue);
+            connectionFinalizerPhantomReferences.add(reference);
+        }
+    }
+
+    private static void finalizeResourceAndRemoveReference(ConnectionFinalizerPhantomReference reference) {
+        try {
+            reference.finalizeResources();
+            reference.clear();
+        } finally {
+            connectionFinalizerPhantomReferences.remove(reference);
+        }
+    }
+
+    private static class ConnectionFinalizerPhantomReference extends PhantomReference<MysqlConnection> {
+        private NetworkResources networkResources;
+
+        ConnectionFinalizerPhantomReference(MysqlConnection conn, NetworkResources networkResources,
+                ReferenceQueue<? super MysqlConnection> refQueue) {
+            super(conn, refQueue);
+            this.networkResources = networkResources;
+        }
+
+        void finalizeResources() {
+            if (this.networkResources != null) {
+                try {
+                    this.networkResources.forceClose();
+                } finally {
+                    this.networkResources = null;
+                }
+            }
+        }
     }
 }