[#12123] fix(catalog): shut down MySQL AbandonedConnectionCleanupThread and broaden ThreadLocal cleanup to prevent Metaspace leak

catalogs/catalog-common/build.gradle.kts

@@ -34,6 +34,9 @@ dependencies {
 
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.junit.jupiter.params)
+  // Used only by TestClassLoaderResourceCleanerUtils to verify the MySQL
+  // AbandonedConnectionCleanupThread is shut down during ClassLoader cleanup.
+  testImplementation(libs.mysql.driver)
 
   testRuntimeOnly(libs.junit.jupiter.engine)
 }

catalogs/catalog-common/src/main/java/org/apache/gravitino/utils/ClassLoaderResourceCleanerUtils.java

@@ -31,9 +31,8 @@
 
 /**
  * Utility class to clean up resources related to a specific class loader to prevent memory leaks.
- * Gravitino will create a new class loader for each catalog and release it when there exist any
- * changes to the catalog. So, it's important to clean up resources related to the class loader to
- * prevent memory leaks.
+ * Gravitino uses isolated class loaders for catalog implementations, and these class loaders must
+ * be properly cleaned up when no longer needed to prevent Metaspace leaks.
  */
 public class ClassLoaderResourceCleanerUtils {
 
@@ -42,7 +41,13 @@ public class ClassLoaderResourceCleanerUtils {
   private ClassLoaderResourceCleanerUtils() {}
 
   /**
-   * Close all resources related to the given class loader to prevent memory leaks.
+   * Close all resources related to the given class loader to prevent memory leaks. This includes
+   * stopping threads, clearing ThreadLocals, shutting down MySQL's
+   * AbandonedConnectionCleanupThread, and releasing various logging and cloud SDK resources.
+   *
+   * <p><b>Warning:</b> This method performs destructive cleanup that affects all code sharing the
+   * ClassLoader. It must only be called when the ClassLoader is about to be destroyed and no
+   * catalog is still using it, i.e. from a catalog's {@code close()} path.
    *
    * @param classLoader the classloader to be closed
    */
@@ -57,12 +62,9 @@ public static void closeClassLoaderResource(ClassLoader classLoader) {
     executeAndCatch(
         ClassLoaderResourceCleanerUtils::closeStatsDataClearerInFileSystem, classLoader);
 
-    // Stop all threads with the current class loader and clear their threadLocal variables for
-    // jetty threads that are loaded by the current class loader.
-    // For example, thread local `threadData` in FileSystem#StatisticsDataCleaner is created
-    // within jetty thread with the current class loader. However, there are clear by
-    // `catalog.close` in ForkJoinPool in CaffeineCache, in this case, the thread local variable
-    // will not be cleared, so we need to clear them manually here.
+    // Stop threads that run under the target class loader and clear ThreadLocals that reference it.
+    // Such ThreadLocals can sit on any thread (Jetty webserver, Caffeine ForkJoinPool,
+    // catalog-cleaner) and keep the ClassLoader from being collected.
     executeAndCatch(
         ClassLoaderResourceCleanerUtils::stopThreadsAndClearThreadLocalVariables, classLoader);
 
@@ -77,6 +79,10 @@ public static void closeClassLoaderResource(ClassLoader classLoader) {
     executeAndCatch(ClassLoaderResourceCleanerUtils::closeResourceInAzure, classLoader);
 
     executeAndCatch(ClassLoaderResourceCleanerUtils::clearShutdownHooks, classLoader);
+
+    executeAndCatch(
+        ClassLoaderResourceCleanerUtils::shutdownMySQLAbandonedConnectionCleanupThread,
+        classLoader);
   }
 
   /**
@@ -146,7 +152,7 @@ private static void stopThreadsAndClearThreadLocalVariables(ClassLoader classLoa
     for (Thread thread : threads) {
       // First clear thread local variables
       clearThreadLocalMap(thread, classLoader);
-      // Close all threads that are using the FilesetCatalogOperations class loader
+      // Stop all threads that are using the target class loader
       if (runningWithClassLoader(thread, classLoader)) {
         LOG.debug("Interrupting thread: {}", thread.getName());
         thread.setContextClassLoader(null);
@@ -178,8 +184,25 @@ private static Thread[] getAllThreads() {
     return threads;
   }
 
-  private static void clearThreadLocalMap(Thread thread, ClassLoader targetClassLoader) {
-    if (thread == null || !thread.getName().startsWith("Gravitino-webserver-")) {
+  @VisibleForTesting
+  static void clearThreadLocalMap(Thread thread, ClassLoader targetClassLoader) {
+    if (thread == null) {
+      return;
+    }
+
+    // Sweep every application thread, not only the Gravitino-webserver-* ones: a ThreadLocal
+    // pointing at the target ClassLoader can live on any thread, such as a Caffeine ForkJoinPool
+    // worker, a catalog-cleaner thread, or a Hadoop daemon. The check below clears only entries
+    // whose value was loaded by the target ClassLoader, so ThreadLocals owned by other
+    // ClassLoaders are left alone.
+    //
+    // Skip threads whose immediate group is the JVM "system" group (Reference Handler, Finalizer,
+    // Signal Dispatcher, and the like): they hold no catalog ClassLoader references, and reflecting
+    // into their ThreadLocals is best avoided. This checks the immediate group only, so threads in
+    // a sub-group of system (such as InnocuousThreadGroup for common-pool workers) are still swept.
+    // That is intended: ForkJoinPool threads can hold catalog ThreadLocals.
+    ThreadGroup group = thread.getThreadGroup();
+    if (group != null && "system".equals(group.getName())) {
       return;
     }
 
@@ -256,7 +279,7 @@ private static void releaseLogFactoryInCommonLogging(ClassLoader currentClassLoa
       LOG.debug("HTrace is not used, skipping release of HTrace LogFactory...");
     }
 
-    // Release the LogFactory for the FilesetCatalogOperations class loader
+    // Release the LogFactory for the target class loader
     Class<?> logFactoryClass =
         Class.forName("org.apache.commons.logging.LogFactory", true, currentClassLoader);
     MethodUtils.invokeStaticMethod(logFactoryClass, "release", currentClassLoader);
@@ -355,6 +378,36 @@ static boolean isOwnedByClassLoader(Class<?> clazz, ClassLoader classLoader) {
     return classLoader != null && clazz.getClassLoader() == classLoader;
   }
 
+  /**
+   * Shutdown MySQL's AbandonedConnectionCleanupThread to prevent it from holding a reference to the
+   * ClassLoader. Unlike simple thread interruption, {@code uncheckedShutdown()} cleans up the
+   * tracked connections map, releasing references to classes loaded by the target ClassLoader.
+   *
+   * @param classLoader the classloader where the MySQL driver is loaded
+   */
+  @VisibleForTesting
+  static void shutdownMySQLAbandonedConnectionCleanupThread(ClassLoader classLoader)
+      throws Exception {
+    // Load with initialize=false: when this class is owned by the target ClassLoader the MySQL
+    // driver has already been used, so the class is initialized and its cleanup thread started.
+    Class<?> clazz =
+        Class.forName("com.mysql.cj.jdbc.AbandonedConnectionCleanupThread", false, classLoader);
+    // uncheckedShutdown() is a JVM-global static that stops the single cleanup thread tied to this
+    // driver class. If the driver was resolved from a parent or shared ClassLoader (for example a
+    // MySQL entity store loaded on the app ClassLoader), stopping it here would break
+    // abandoned-connection cleanup for the whole JVM. Only act when the target ClassLoader owns the
+    // class, the same guard the other cleanup steps in this class use.
+    if (!isOwnedByClassLoader(clazz, classLoader)) {
+      LOG.debug(
+          "AbandonedConnectionCleanupThread is owned by {}, not the target classloader {}; skipping shutdown",
+          clazz.getClassLoader(),
+          classLoader);
+      return;
+    }
+    MethodUtils.invokeStaticMethod(clazz, "uncheckedShutdown");
+    LOG.info("AbandonedConnectionCleanupThread has been shutdown.");
+  }
+
   @FunctionalInterface
   private interface ThrowableConsumer<T> {
     void accept(T t) throws Exception;