[#12986] fix(catalog): Release the ClassLoader of a dropped catalog

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

@@ -20,10 +20,20 @@
 package org.apache.gravitino.utils;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.lang.ref.Reference;
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.security.Provider;
+import java.security.Security;
+import java.util.Collection;
 import java.util.IdentityHashMap;
+import java.util.ResourceBundle;
 import java.util.Timer;
 import java.util.concurrent.ScheduledExecutorService;
+import javax.annotation.Nullable;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.commons.lang3.reflect.MethodUtils;
 import org.slf4j.Logger;
@@ -70,6 +80,16 @@ public static void closeClassLoaderResource(ClassLoader classLoader) {
     // instance.
     executeAndCatch(ClassLoaderResourceCleanerUtils::releaseLogFactoryInCommonLogging, classLoader);
 
+    executeAndCatch(ClassLoaderResourceCleanerUtils::removeLoggerContextListeners, classLoader);
+
+    executeAndCatch(ClassLoaderResourceCleanerUtils::deregisterJdbcDrivers, classLoader);
+
+    executeAndCatch(ClassLoaderResourceCleanerUtils::shutdownMysqlConnectionCleanup, classLoader);
+
+    executeAndCatch(ClassLoaderResourceCleanerUtils::removeSecurityProviders, classLoader);
+
+    executeAndCatch(ClassLoaderResourceCleanerUtils::clearResourceBundleCache, classLoader);
+
     executeAndCatch(ClassLoaderResourceCleanerUtils::closeResourceInAWS, classLoader);
 
     executeAndCatch(ClassLoaderResourceCleanerUtils::closeResourceInGCP, classLoader);
@@ -160,8 +180,37 @@ private static void stopThreadsAndClearThreadLocalVariables(ClassLoader classLoa
     }
   }
 
-  private static boolean runningWithClassLoader(Thread thread, ClassLoader targetClassLoader) {
-    return thread != null && thread.getContextClassLoader() == targetClassLoader;
+  /**
+   * Whether the thread belongs to the class loader being released.
+   *
+   * <p>The context ClassLoader is only one of the ways a thread can carry a catalog. A driver that
+   * starts its own housekeeping thread, such as PostgreSQL's {@code LazyCleaner}, is running code
+   * defined by the catalog's loader: the thread is a GC root, so its class alone keeps the loader
+   * alive no matter what its context ClassLoader says.
+   *
+   * <p>Ownership has to be read from the thread itself, never from what it happens to be running: a
+   * request thread executing an operation on this very catalog is not the catalog's to stop, and
+   * interrupting it fails the request with "Thread was interrupted while waiting for lock".
+   */
+  @VisibleForTesting
+  static boolean runningWithClassLoader(Thread thread, ClassLoader targetClassLoader) {
+    if (thread == null) {
+      return false;
+    }
+    if (thread.getContextClassLoader() == targetClassLoader
+        || thread.getClass().getClassLoader() == targetClassLoader) {
+      return true;
+    }
+    try {
+      Object runnable = FieldUtils.readField(thread, "target", true);
+      if (runnable != null && runnable.getClass().getClassLoader() == targetClassLoader) {
+        return true;
+      }
+    } catch (Exception e) {
+      LOG.debug("Cannot read the runnable of thread {}", thread.getName(), e);
+    }
+
+    return false;
   }
 
   private static Thread[] getAllThreads() {
@@ -178,8 +227,9 @@ 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;
     }
 
@@ -197,9 +247,10 @@ private static void clearThreadLocalMap(Thread thread, ClassLoader targetClassLo
         for (Object entry : table) {
           if (entry != null) {
             Object value = FieldUtils.readField(entry, "value", true);
-            if (value != null
-                && value.getClass().getClassLoader() != null
-                && value.getClass().getClassLoader() == targetClassLoader) {
+            // The entry is a WeakReference to the ThreadLocal itself, which can be the leaking
+            // side when the ThreadLocal was declared by a class of the dying catalog.
+            Object key = entry instanceof Reference ? ((Reference<?>) entry).get() : null;
+            if (definedBy(value, targetClassLoader) || definedBy(key, targetClassLoader)) {
               LOG.debug(
                   "Cleaning up thread local {} for thread {} with custom class loader",
                   value,
@@ -214,6 +265,31 @@ private static void clearThreadLocalMap(Thread thread, ClassLoader targetClassLo
     }
   }
 
+  /**
+   * Whether {@code value}, or what it refers to when it is a {@link Reference}, was defined by
+   * {@code classLoader}.
+   *
+   * <p>Looking through a {@link Reference} matters: caches such as Jackson's {@code BufferRecycler}
+   * park a {@code SoftReference} in a {@link ThreadLocal}. The reference itself is a bootstrap
+   * class, so only its referent identifies the owning catalog. Left in place, such an entry keeps
+   * the catalog's ClassLoader alive until heap pressure clears the soft reference, which Metaspace
+   * pressure alone never triggers.
+   */
+  @VisibleForTesting
+  static boolean definedBy(@Nullable Object value, ClassLoader classLoader) {
+    if (value == null) {
+      return false;
+    }
+    if (value.getClass().getClassLoader() == classLoader) {
+      return true;
+    }
+    if (value instanceof Reference) {
+      Object referent = ((Reference<?>) value).get();
+      return referent != null && referent.getClass().getClassLoader() == classLoader;
+    }
+    return false;
+  }
+
   /**
    * Clear shutdown hooks registered by the target class loader to prevent memory leaks.
    *
@@ -236,6 +312,145 @@ private static void clearShutdownHooks(ClassLoader targetClassLoader) throws Exc
             });
   }
 
+  /**
+   * Removes shutdown listeners the class loader registered on the shared Log4j {@code
+   * LoggerContext}.
+   *
+   * <p>commons-logging's {@code Log4jApiLogFactory} registers a {@code LogAdapter} with the
+   * LoggerContext of the server, which outlives every catalog. {@code LogFactory.release} drops the
+   * factory from its own cache but leaves that registration in place, so the adapter's class, and
+   * through it the catalog's ClassLoader, stays reachable from a static for the life of the
+   * process.
+   */
+  /**
+   * Drops the {@link ResourceBundle} cache entries loaded through this class loader.
+   *
+   * <p>{@link ResourceBundle} caches bundles in a JVM-wide static map, behind soft references. A
+   * driver that loads message bundles, such as Oracle's {@code ErrorMessages}, therefore leaves its
+   * class - and the catalog's ClassLoader - reachable until heap pressure clears the soft
+   * reference, which Metaspace pressure alone never causes.
+   */
+  @VisibleForTesting
+  static void clearResourceBundleCache(ClassLoader targetClassLoader) {
+    ResourceBundle.clearCache(targetClassLoader);
+  }
+
+  /**
+   * Removes the JCA security providers the class loader installed.
+   *
+   * <p>{@link Security} keeps installed providers in a JVM-wide static list. Hadoop's cloud
+   * connectors install one, such as the shaded {@code OpenSSLProvider} that ships in the AWS
+   * bundle, and it is never removed, so the provider's class holds the catalog's loader for the
+   * life of the process.
+   */
+  @VisibleForTesting
+  static void removeSecurityProviders(ClassLoader targetClassLoader) {
+    for (Provider provider : Security.getProviders()) {
+      if (provider.getClass().getClassLoader() == targetClassLoader) {
+        Security.removeProvider(provider.getName());
+        LOG.info("Removed security provider {} of a released catalog ClassLoader", provider);
+      }
+    }
+  }
+
+  /**
+   * Shuts down MySQL Connector/J's abandoned-connection cleanup thread when the driver belongs to
+   * this class loader.
+   *
+   * <p>The driver keeps that thread and its executor in a static field, and the executor's thread
+   * factory is a lambda defined by the catalog's loader, so a running cleanup thread pins the
+   * loader through its own stack frame. Connector/J exposes {@code uncheckedShutdown()} for exactly
+   * this case.
+   */
+  private static void shutdownMysqlConnectionCleanup(ClassLoader targetClassLoader)
+      throws Exception {
+    Class<?> cleanupThreadClass =
+        Class.forName(
+            "com.mysql.cj.jdbc.AbandonedConnectionCleanupThread", true, targetClassLoader);
+    if (!isOwnedByClassLoader(cleanupThreadClass, targetClassLoader)) {
+      LOG.debug(
+          "MySQL Connector/J is owned by {}, not {}; skipping shared-class cleanup",
+          cleanupThreadClass.getClassLoader(),
+          targetClassLoader);
+      return;
+    }
+    // uncheckedShutdown stops the thread even when the driver still believes it is in use, which
+    // is what unloading the ClassLoader requires; checkedShutdown returns without doing anything.
+    MethodUtils.invokeStaticMethod(cleanupThreadClass, "uncheckedShutdown");
+    LOG.info("Shut down the MySQL abandoned-connection cleanup thread of a released ClassLoader");
+  }
+
+  /**
+   * Deregisters the JDBC drivers the class loader registered with {@link java.sql.DriverManager}.
+   *
+   * <p>{@code DriverManager} keeps registered drivers in a static list, and a driver defined by a
+   * catalog's ClassLoader keeps that loader alive for the life of the process. It cannot be removed
+   * from here directly: {@code DriverManager} filters both {@code getDrivers()} and {@code
+   * deregisterDriver()} by the class loader of the calling class, so from the server's ClassLoader
+   * the catalog's drivers are not even visible. Defining {@link JdbcDriverDeregisterer} inside the
+   * target loader and calling it there gives {@code DriverManager} a caller that owns them.
+   */
+  @VisibleForTesting
+  static void deregisterJdbcDrivers(ClassLoader targetClassLoader) throws Exception {
+    String name = JdbcDriverDeregisterer.class.getName();
+    byte[] bytecode;
+    try (InputStream in =
+        ClassLoaderResourceCleanerUtils.class
+            .getClassLoader()
+            .getResourceAsStream(name.replace('.', '/') + ".class")) {
+      if (in == null) {
+        LOG.debug("Cannot locate the bytecode of {}, skipping JDBC driver cleanup", name);
+        return;
+      }
+      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+      byte[] chunk = new byte[8192];
+      int read;
+      while ((read = in.read(chunk)) != -1) {
+        buffer.write(chunk, 0, read);
+      }
+      bytecode = buffer.toByteArray();
+    }
+
+    Method defineClass =
+        ClassLoader.class.getDeclaredMethod(
+            "defineClass", String.class, byte[].class, int.class, int.class);
+    defineClass.setAccessible(true);
+    Class<?> deregisterer;
+    try {
+      deregisterer =
+          (Class<?>) defineClass.invoke(targetClassLoader, name, bytecode, 0, bytecode.length);
+    } catch (InvocationTargetException e) {
+      if (e.getCause() instanceof LinkageError) {
+        // Already defined by an earlier cleanup of the same loader, whose drivers are gone.
+        LOG.debug("{} is already defined in {}", name, targetClassLoader);
+        return;
+      }
+      throw e;
+    }
+
+    Object deregistered = MethodUtils.invokeStaticMethod(deregisterer, "deregisterAll");
+    if (deregistered instanceof Collection && !((Collection<?>) deregistered).isEmpty()) {
+      LOG.info("Deregistered JDBC driver(s) {} of a released catalog ClassLoader", deregistered);
+    }
+  }
+
+  @VisibleForTesting
+  static void removeLoggerContextListeners(ClassLoader targetClassLoader) throws Exception {
+    Class<?> logManagerClass = Class.forName("org.apache.logging.log4j.LogManager");
+    Object contextFactory = MethodUtils.invokeStaticMethod(logManagerClass, "getFactory");
+    Object selector = MethodUtils.invokeMethod(contextFactory, "getSelector");
+    Collection<?> contexts =
+        (Collection<?>) MethodUtils.invokeMethod(selector, "getLoggerContexts");
+    for (Object context : contexts) {
+      Collection<?> listeners = (Collection<?>) FieldUtils.readField(context, "listeners", true);
+      if (listeners != null) {
+        listeners.removeIf(
+            listener ->
+                listener != null && listener.getClass().getClassLoader() == targetClassLoader);
+      }
+    }
+  }
+
   /**
    * Release the LogFactory for the target class loader to prevent memory leaks.
    *

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

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.utils;
+
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.List;
+
+/**
+ * Deregisters the JDBC drivers a catalog's ClassLoader registered with {@link DriverManager}.
+ *
+ * <p>This class is never called through its own name. {@link DriverManager} filters both {@code
+ * getDrivers()} and {@code deregisterDriver()} by the class loader of the calling class, so a
+ * driver defined by a catalog's isolated ClassLoader is invisible, and undeletable, from the server
+ * ClassLoader. {@link ClassLoaderResourceCleanerUtils} therefore defines a copy of this class
+ * inside the catalog's ClassLoader and invokes it reflectively, so that {@code DriverManager} sees
+ * a caller that owns the drivers. Keep its dependencies to {@code java.*} only: the copy is defined
+ * directly from bytecode and resolves everything through the catalog's ClassLoader.
+ */
+public final class JdbcDriverDeregisterer {
+
+  private JdbcDriverDeregisterer() {}
+
+  /**
+   * Deregisters every driver defined by the ClassLoader of this class.
+   *
+   * @return the names of the drivers that were deregistered
+   */
+  public static List<String> deregisterAll() {
+    ClassLoader owner = JdbcDriverDeregisterer.class.getClassLoader();
+    List<String> deregistered = new ArrayList<>();
+    Enumeration<Driver> drivers = DriverManager.getDrivers();
+    while (drivers.hasMoreElements()) {
+      Driver driver = drivers.nextElement();
+      if (driver.getClass().getClassLoader() == owner) {
+        try {
+          DriverManager.deregisterDriver(driver);
+          deregistered.add(driver.getClass().getName());
+        } catch (Exception e) {
+          // Leave the driver registered rather than failing the whole cleanup.
+        }
+      }
+    }
+    return deregistered;
+  }
+}