This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new f31c2a0b11 [#11301] fix(catalog): guard
ClassLoaderResourceCleanerUtils against shared-class static-field mutation
(#11431)
f31c2a0b11 is described below
commit f31c2a0b11e5743dce185405011e432e787989ae
Author: Qi Yu <[email protected]>
AuthorDate: Fri Jun 5 14:11:51 2026 +0800
[#11301] fix(catalog): guard ClassLoaderResourceCleanerUtils against
shared-class static-field mutation (#11431)
### What changes were proposed in this pull request?
When `IsolatedClassLoader` resolves Hadoop/AWS/Azure classes via parent
delegation (e.g. `hadoop-common.jar` or `gravitino-aws-bundle.jar`
appears
on the AppClassLoader's classpath), `Class.forName(...,
targetClassLoader)`
returns the parent-loaded class. The static fields on that class
(`MutableQuantiles.scheduler`, `FileSystem.CACHE`, `AwsSdkMetrics`
MBean,
`AbfsClientThrottlingIntercept` timers) are JVM-global and shared across
every catalog. Calling `scheduler.shutdownNow()` on the global instance
permanently terminates Hadoop metrics scheduling, causing
`RejectedExecutionException` on any subsequent catalog that creates a
`MutableQuantiles` instance (e.g. S3A FileSystem instrumentation).
This PR adds `isOwnedByClassLoader(clazz, targetClassLoader)` guards
before
any static-state mutation in `ClassLoaderResourceCleanerUtils`. If the
class
was delegated to a parent ClassLoader, cleanup is skipped; if it belongs
to the
catalog's own ClassLoader (the normal isolated case), cleanup proceeds
as before.
The same guard is applied to:
- `closeStatsDataClearerInFileSystem`: `FileSystem.closeAll()`,
scheduler shutdown, `STATS_DATA_CLEANER` thread interrupt.
- `closeResourceInAWS`: `AwsSdkMetrics` MBean unregistration.
- `closeResourceInGCP`: shaded `LogFactory` release.
- `closeResourceInAzure`: `AbfsClientThrottlingIntercept` timer
cancellation.
Additionally, `closeClassLoaderResource` calls are added to
`HiveCatalogOperations` and `FilesetCatalogOperations`, which use Hadoop
FileSystem extensively but had no classloader cleanup on close, risking
thread
and memory leaks.
### Why are the changes needed?
Fix: #11301
After an Iceberg or Paimon catalog is evicted from the cache and
re-created,
any subsequent S3A FileSystem initialization fails with:
```
RejectedExecutionException: Task ... rejected from
ScheduledThreadPoolExecutor@...[Terminated, pool size = 0, active threads =
0,
queued tasks = 0, completed tasks = 23664]
```
because `closeStatsDataClearerInFileSystem` shut down the JVM-global
`MutableQuantiles.scheduler` when Hadoop happened to be on the
AppClassLoader's
classpath.
### Does this PR introduce _any_ user-facing change?
No functional behavior change for correctly isolated deployments. For
deployments
where Hadoop is on the server's main classpath, the broken
scheduler-shutdown is
now skipped, restoring correct behavior after catalog re-creation.
### How was this patch tested?
- `./gradlew :catalogs:catalog-common:spotlessApply
:catalogs:catalog-hive:spotlessApply
:catalogs:catalog-fileset:spotlessApply`
- `./gradlew :catalogs:catalog-hive:test :catalogs:catalog-fileset:test
-PskipDockerTests=true`
---------
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../utils/ClassLoaderResourceCleanerUtils.java | 84 ++++++++++++++++++----
.../utils/TestClassLoaderResourceCleanerUtils.java | 71 ++++++++++++++++++
.../catalog/fileset/FilesetCatalogOperations.java | 3 +
.../catalog/hive/HiveCatalogOperations.java | 2 +
4 files changed, 145 insertions(+), 15 deletions(-)
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/utils/ClassLoaderResourceCleanerUtils.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/utils/ClassLoaderResourceCleanerUtils.java
index 9ecece6eec..6a91f4ac48 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/utils/ClassLoaderResourceCleanerUtils.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/utils/ClassLoaderResourceCleanerUtils.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.utils;
+import com.google.common.annotations.VisibleForTesting;
import java.lang.reflect.Field;
import java.util.IdentityHashMap;
import java.util.Timer;
@@ -87,6 +88,21 @@ public class ClassLoaderResourceCleanerUtils {
throws Exception {
Class<?> fileSystemClass =
Class.forName("org.apache.hadoop.fs.FileSystem", true,
targetClassLoader);
+
+ // If FileSystem was resolved from a parent/AppClassLoader rather than the
catalog's own
+ // classloader, its CACHE, Statistics cleaner, and MutableQuantiles
scheduler are shared
+ // across all catalogs in the JVM. Operating on shared static state here
would close every
+ // catalog's FileSystems and permanently terminate the global scheduler,
breaking any
+ // subsequent catalog that uses Hadoop metrics. Skip cleanup for shared
classes and let
+ // the JVM manage them.
+ if (!isOwnedByClassLoader(fileSystemClass, targetClassLoader)) {
+ LOG.debug(
+ "Hadoop FileSystem is owned by {}, not the target classloader {};
skipping shared-class cleanup",
+ fileSystemClass.getClassLoader(),
+ targetClassLoader);
+ return;
+ }
+
MethodUtils.invokeStaticMethod(fileSystemClass, "closeAll");
Class<?> mutableQuantilesClass =
@@ -94,16 +110,22 @@ public class ClassLoaderResourceCleanerUtils {
Class<?> statisticsClass =
Class.forName("org.apache.hadoop.fs.FileSystem$Statistics", true,
targetClassLoader);
- ScheduledExecutorService scheduler =
- (ScheduledExecutorService)
- FieldUtils.readStaticField(mutableQuantilesClass, "scheduler",
true);
- scheduler.shutdownNow();
- Field statisticsCleanerField = FieldUtils.getField(statisticsClass,
"STATS_DATA_CLEANER", true);
- Object statisticsCleaner = statisticsCleanerField.get(null);
- if (statisticsCleaner != null) {
- ((Thread) statisticsCleaner).interrupt();
- ((Thread) statisticsCleaner).setContextClassLoader(null);
- ((Thread) statisticsCleaner).join();
+ if (isOwnedByClassLoader(mutableQuantilesClass, targetClassLoader)) {
+ ScheduledExecutorService scheduler =
+ (ScheduledExecutorService)
+ FieldUtils.readStaticField(mutableQuantilesClass, "scheduler",
true);
+ scheduler.shutdownNow();
+ }
+
+ if (isOwnedByClassLoader(statisticsClass, targetClassLoader)) {
+ Field statisticsCleanerField =
+ FieldUtils.getField(statisticsClass, "STATS_DATA_CLEANER", true);
+ Object statisticsCleaner = statisticsCleanerField.get(null);
+ if (statisticsCleaner != null) {
+ ((Thread) statisticsCleaner).interrupt();
+ ((Thread) statisticsCleaner).setContextClassLoader(null);
+ ((Thread) statisticsCleaner).join();
+ }
}
}
@@ -246,19 +268,35 @@ public class ClassLoaderResourceCleanerUtils {
* @param classLoader the classloader where AWS SDK is loaded
*/
private static void closeResourceInAWS(ClassLoader classLoader) throws
Exception {
- // For Aws SDK metrics, unregister the metric admin MBean
Class<?> awsSdkMetricsClass =
Class.forName("com.amazonaws.metrics.AwsSdkMetrics", true,
classLoader);
+ // AwsSdkMetrics holds a static MBeanServer registration. If the class was
delegated to a
+ // parent/AppClassLoader, unregistering here would remove the MBean for
the entire JVM.
+ if (!isOwnedByClassLoader(awsSdkMetricsClass, classLoader)) {
+ LOG.debug(
+ "AwsSdkMetrics is owned by {}, not {}; skipping MBean unregister",
+ awsSdkMetricsClass.getClassLoader(),
+ classLoader);
+ return;
+ }
MethodUtils.invokeStaticMethod(awsSdkMetricsClass,
"unregisterMetricAdminMBean");
}
private static void closeResourceInGCP(ClassLoader classLoader) throws
Exception {
- // For GCS
Class<?> relocatedLogFactory =
Class.forName(
"org.apache.gravitino.gcp.shaded.org.apache.commons.logging.LogFactory",
true,
classLoader);
+ // The GCP shaded LogFactory is always bundled inside the GCP plugin; if
it resolves to a
+ // different classloader, skip to avoid releasing a shared factory.
+ if (!isOwnedByClassLoader(relocatedLogFactory, classLoader)) {
+ LOG.debug(
+ "GCP shaded LogFactory is owned by {}, not {}; skipping release",
+ relocatedLogFactory.getClassLoader(),
+ classLoader);
+ return;
+ }
MethodUtils.invokeStaticMethod(relocatedLogFactory, "release",
classLoader);
}
@@ -272,12 +310,20 @@ public class ClassLoaderResourceCleanerUtils {
* @param classLoader the classloader where Azure Blob File System is loaded
*/
private static void closeResourceInAzure(ClassLoader classLoader) throws
Exception {
- // Clear timer in AbfsClientThrottlingAnalyzer
Class<?> abfsClientThrottlingInterceptClass =
Class.forName(
"org.apache.hadoop.fs.azurebfs.services.AbfsClientThrottlingIntercept",
true,
classLoader);
+ // AbfsClientThrottlingIntercept holds a static singleton with Timers. If
the ABFS class was
+ // delegated to a parent/AppClassLoader, cancelling its timers would break
ABFS for the JVM.
+ if (!isOwnedByClassLoader(abfsClientThrottlingInterceptClass,
classLoader)) {
+ LOG.debug(
+ "AbfsClientThrottlingIntercept is owned by {}, not {}; skipping
Azure cleanup",
+ abfsClientThrottlingInterceptClass.getClassLoader(),
+ classLoader);
+ return;
+ }
Object abfsClientThrottlingIntercept =
FieldUtils.readStaticField(abfsClientThrottlingInterceptClass,
"singleton", true);
@@ -291,8 +337,6 @@ public class ClassLoaderResourceCleanerUtils {
Timer writeTimer = (Timer) FieldUtils.readField(writeThrottler, "timer",
true);
writeTimer.cancel();
- // Release the LogFactory for the Azure shaded commons logging which has
been relocated
- // by the Azure SDK
Class<?> relocatedLogFactory =
Class.forName(
"org.apache.gravitino.azure.shaded.org.apache.commons.logging.LogFactory",
@@ -301,6 +345,16 @@ public class ClassLoaderResourceCleanerUtils {
MethodUtils.invokeStaticMethod(relocatedLogFactory, "release",
classLoader);
}
+ /**
+ * Returns true if {@code clazz} was loaded directly by {@code classLoader}
(not delegated to a
+ * parent). Use this before touching static fields that must belong to the
catalog's own
+ * classloader to avoid accidentally mutating JVM-global shared state.
+ */
+ @VisibleForTesting
+ static boolean isOwnedByClassLoader(Class<?> clazz, ClassLoader classLoader)
{
+ return classLoader != null && clazz.getClassLoader() == classLoader;
+ }
+
@FunctionalInterface
private interface ThrowableConsumer<T> {
void accept(T t) throws Exception;
diff --git
a/catalogs/catalog-common/src/test/java/org/apache/gravitino/utils/TestClassLoaderResourceCleanerUtils.java
b/catalogs/catalog-common/src/test/java/org/apache/gravitino/utils/TestClassLoaderResourceCleanerUtils.java
new file mode 100644
index 0000000000..c5a241d8c7
--- /dev/null
+++
b/catalogs/catalog-common/src/test/java/org/apache/gravitino/utils/TestClassLoaderResourceCleanerUtils.java
@@ -0,0 +1,71 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import org.junit.jupiter.api.Test;
+
+class TestClassLoaderResourceCleanerUtils {
+
+ /**
+ * When a class is loaded by exactly the target classloader,
isOwnedByClassLoader must return true
+ * — the guard should allow static-state cleanup to proceed.
+ */
+ @Test
+ void testIsOwnedByClassLoaderReturnsTrueForOwningLoader() {
+ ClassLoader loader =
ClassLoaderResourceCleanerUtils.class.getClassLoader();
+ assertTrue(
+ ClassLoaderResourceCleanerUtils.isOwnedByClassLoader(
+ ClassLoaderResourceCleanerUtils.class, loader));
+ }
+
+ /**
+ * When a class was resolved via parent delegation (i.e. the actual loader
is the parent, not the
+ * child), isOwnedByClassLoader must return false — the guard should skip
cleanup to avoid
+ * mutating shared JVM-global static state.
+ */
+ @Test
+ void testIsOwnedByClassLoaderReturnsFalseForParentDelegatedClass() throws
Exception {
+ ClassLoader parent =
ClassLoaderResourceCleanerUtils.class.getClassLoader();
+ // Child delegates everything to the parent;
ClassLoaderResourceCleanerUtils is therefore
+ // parent-loaded, not child-loaded.
+ try (URLClassLoader child = new URLClassLoader(new URL[0], parent)) {
+ assertFalse(
+ ClassLoaderResourceCleanerUtils.isOwnedByClassLoader(
+ ClassLoaderResourceCleanerUtils.class, child));
+ }
+ }
+
+ /**
+ * Bootstrap-loaded classes (whose getClassLoader() returns null) are never
"owned" by a named
+ * classloader — the guard must return false for them too.
+ */
+ @Test
+ void testIsOwnedByClassLoaderReturnsFalseForBootstrapLoadedClass() {
+ // String is loaded by the bootstrap classloader; getClassLoader() returns
null.
+ assertFalse(
+ ClassLoaderResourceCleanerUtils.isOwnedByClassLoader(
+ String.class, ClassLoader.getSystemClassLoader()));
+ }
+}
diff --git
a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
index ed660b850f..4b86c37fd6 100644
---
a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
+++
b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
@@ -104,6 +104,7 @@ import org.apache.gravitino.meta.FilesetEntity;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.metrics.MetricsSystem;
import org.apache.gravitino.metrics.source.FilesetCatalogMetricsSource;
+import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils;
import org.apache.gravitino.utils.FilesetUtil;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
@@ -985,6 +986,8 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
if (metricsSystem != null) {
metricsSystem.unregister(catalogMetricsSource);
}
+
+
ClassLoaderResourceCleanerUtils.closeClassLoaderResource(this.getClass().getClassLoader());
}
private void validateLocationHierarchy(
diff --git
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
index 87825ca913..1a474d4f53 100644
---
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
+++
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
@@ -89,6 +89,7 @@ import
org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.expressions.transforms.Transforms;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.types.Type;
+import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils;
import org.apache.gravitino.utils.PrincipalUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -196,6 +197,7 @@ public class HiveCatalogOperations
clientPool.close();
clientPool = null;
}
+
ClassLoaderResourceCleanerUtils.closeClassLoaderResource(this.getClass().getClassLoader());
}
/**
