[#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,13 @@
 package org.apache.gravitino.utils;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.lang.ref.Reference;
 import java.lang.reflect.Field;
+import java.util.Collection;
 import java.util.IdentityHashMap;
 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 +73,8 @@ public static void closeClassLoaderResource(ClassLoader classLoader) {
     // instance.
     executeAndCatch(ClassLoaderResourceCleanerUtils::releaseLogFactoryInCommonLogging, classLoader);
 
+    executeAndCatch(ClassLoaderResourceCleanerUtils::removeLoggerContextListeners, classLoader);
+
     executeAndCatch(ClassLoaderResourceCleanerUtils::closeResourceInAWS, classLoader);
 
     executeAndCatch(ClassLoaderResourceCleanerUtils::closeResourceInGCP, classLoader);
@@ -178,8 +183,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 +203,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 +221,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 +268,33 @@ 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.
+   */
+  @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/test/java/org/apache/gravitino/utils/TestClassLoaderResourceCleanerUtils.java

@@ -20,14 +20,92 @@
 package org.apache.gravitino.utils;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.lang.ref.SoftReference;
+import java.lang.ref.WeakReference;
 import java.net.URL;
 import java.net.URLClassLoader;
 import org.junit.jupiter.api.Test;
 
 class TestClassLoaderResourceCleanerUtils {
 
+  private static final ThreadLocal<Object> SOFT_HOLDER = new ThreadLocal<>();
+  private static final ThreadLocal<Object> UNRELATED_HOLDER = new ThreadLocal<>();
+
+  /** A class with no dependencies beyond java.*, so a bare-bones child loader can define it. */
+  public static class Leaky {}
+
+  private static URLClassLoader childLoaderOwning(Class<?> clazz) throws Exception {
+    URL location = clazz.getProtectionDomain().getCodeSource().getLocation();
+    // A null parent keeps delegation off the app loader, so the child defines the class itself.
+    return new URLClassLoader(new URL[] {location}, null);
+  }
+
+  /** The value's own class identifies the owner in the simple case. */
+  @Test
+  void testDefinedByMatchesTheDeclaringLoader() throws Exception {
+    try (URLClassLoader child = childLoaderOwning(Leaky.class)) {
+      Object leaky = child.loadClass(Leaky.class.getName()).getDeclaredConstructor().newInstance();
+      assertTrue(ClassLoaderResourceCleanerUtils.definedBy(leaky, child));
+      assertFalse(ClassLoaderResourceCleanerUtils.definedBy(leaky, Leaky.class.getClassLoader()));
+    }
+  }
+
+  /**
+   * Caches such as Jackson's BufferRecycler park a SoftReference in a ThreadLocal. The reference is
+   * a bootstrap class, so only its referent identifies the owning catalog.
+   */
+  @Test
+  void testDefinedByLooksThroughAReference() throws Exception {
+    try (URLClassLoader child = childLoaderOwning(Leaky.class)) {
+      Object leaky = child.loadClass(Leaky.class.getName()).getDeclaredConstructor().newInstance();
+      assertTrue(ClassLoaderResourceCleanerUtils.definedBy(new SoftReference<>(leaky), child));
+      assertTrue(ClassLoaderResourceCleanerUtils.definedBy(new WeakReference<>(leaky), child));
+    }
+  }
+
+  /** An empty reference names no owner and must not be mistaken for one. */
+  @Test
+  void testDefinedByIgnoresNullAndClearedReferences() {
+    assertFalse(ClassLoaderResourceCleanerUtils.definedBy(null, getClass().getClassLoader()));
+    assertFalse(
+        ClassLoaderResourceCleanerUtils.definedBy(
+            new SoftReference<>(null), getClass().getClassLoader()));
+  }
+
+  /**
+   * A thread local holding the catalog's object behind a SoftReference must be cleared. Left in
+   * place it keeps the catalog's ClassLoader alive until heap pressure clears the reference, which
+   * Metaspace pressure alone never triggers.
+   */
+  @Test
+  void testClearThreadLocalMapClearsSoftReferencedValues() throws Exception {
+    try (URLClassLoader child = childLoaderOwning(Leaky.class)) {
+      Object leaky = child.loadClass(Leaky.class.getName()).getDeclaredConstructor().newInstance();
+      SOFT_HOLDER.set(new SoftReference<>(leaky));
+
+      ClassLoaderResourceCleanerUtils.clearThreadLocalMap(Thread.currentThread(), child);
+
+      assertNull(SOFT_HOLDER.get());
+    }
+  }
+
+  /** Entries belonging to another loader must survive the sweep. */
+  @Test
+  void testClearThreadLocalMapLeavesUnrelatedValues() throws Exception {
+    try (URLClassLoader child = childLoaderOwning(Leaky.class)) {
+      Object unrelated = new Object();
+      UNRELATED_HOLDER.set(unrelated);
+
+      ClassLoaderResourceCleanerUtils.clearThreadLocalMap(Thread.currentThread(), child);
+
+      assertSame(unrelated, UNRELATED_HOLDER.get());
+    }
+  }
+
   /**
    * When a class is loaded by exactly the target classloader, isOwnedByClassLoader must return true
    * — the guard should allow static-state cleanup to proceed.

catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/client/HiveClientFactory.java

@@ -33,6 +33,7 @@
 import org.apache.gravitino.exceptions.GravitinoRuntimeException;
 import org.apache.gravitino.hive.kerberos.AuthenticationConfig;
 import org.apache.gravitino.hive.kerberos.HmsKerberosClient;
+import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils;
 import org.apache.gravitino.utils.PrincipalUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -260,6 +261,12 @@ public void close() {
 
       synchronized (classLoaderLock) {
         if (backendClassLoader != null) {
+          // The backend ClassLoader is a second, nested isolation layer that holds the catalog's
+          // own ClassLoader as its base. Closing it releases its jars but not the references other
+          // threads still hold to it: Hadoop's Shell runs sub-processes, and the JDK's pooled
+          // "process reaper" threads inherit the spawning thread's context ClassLoader, which is a
+          // GC root. Cleaning the nested loader clears those, so both layers become collectable.
+          ClassLoaderResourceCleanerUtils.closeClassLoaderResource(backendClassLoader);
           backendClassLoader.close();
           backendClassLoader = null;
         }