This is an automated email from the ASF dual-hosted git repository.
yuqi1129 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 d9c9237410 [#10093] feat(core): Introduce ClassLoaderPool to share
ClassLoaders across same-type catalogs (#10480)
d9c9237410 is described below
commit d9c9237410259ea3f396ef4ef207a3ab2312666f
Author: YangJie <[email protected]>
AuthorDate: Tue Jul 21 16:00:36 2026 +0800
[#10093] feat(core): Introduce ClassLoaderPool to share ClassLoaders across
same-type catalogs (#10480)
### What changes were proposed in this pull request?
Introduce a `ClassLoaderPool` with reference counting to share
`IsolatedClassLoader` instances across catalogs of the same type, and
centralize ClassLoader resource cleanup into the pool's lifecycle. Add
`gravitino.catalog.classloader.sharing.enabled` as a safety-net toggle
to disable sharing if needed.
**Core mechanism:** Catalogs with identical isolation-relevant
properties share a single `IsolatedClassLoader`. The isolation key is
built from the provider plus the package, authorization provider,
authentication type, Kerberos principal/keytab, and backend URIs
(`metastore.uris`, `jdbc-url`, `uri`, `fs.defaultFS`). The pool uses
`ConcurrentHashMap.compute()` for atomic acquire/release, and runs
cleanup (JDBC driver deregistration, then the shared
`ClassLoaderResourceCleanerUtils` pass over Hadoop FileSystem,
ThreadLocals, commons-logging, cloud SDKs, and shutdown hooks) only when
the last catalog releases the shared ClassLoader.
When sharing is disabled via
`gravitino.catalog.classloader.sharing.enabled=false`, each catalog gets
its own dedicated ClassLoader without going through the pool — restoring
the pre-PR behavior as a fallback.
**New classes:**
- `ClassLoaderKey` — `Map<String, String>`-based key for ClassLoader
sharing, decoupled from specific property names
- `ClassLoaderPool` — thread-safe pool with reference counting and
lifecycle management
- `PooledClassLoaderEntry` — holds a shared ClassLoader and its
reference count
**Changes to existing classes:**
- `CatalogManager` — integrates the pool into catalog creation, test
connection, and close paths; adds a non-pooled path gated by
`classLoaderSharingEnabled`; extracts `initCatalogWrapper()` to remove
duplication between the pooled and non-pooled paths; fixes ClassLoader
leaks in `testConnection()` and `getResolvedProperties()`
- `Configs` — adds `gravitino.catalog.classloader.sharing.enabled`
- `IsolatedClassLoader` — exposes `getInternalClassLoader()` so the pool
can run cleanup on the underlying `URLClassLoader`
- Removes per-catalog cleanup calls from `JdbcCatalogOperations`,
`IcebergCatalogOperations`, `IcebergCatalogWrapper`,
`PaimonCatalogOperations`, `FilesetCatalogOperations`,
`HiveCatalogOperations`, and `CatalogWrapperForREST`, since cleanup now
happens once in the pool on final release
### Why are the changes needed?
Concurrent catalog creation with different names but the same provider
type causes `OutOfMemoryError: Metaspace`. Each catalog creates an
independent `IsolatedClassLoader` that loads all provider JARs into
Metaspace. With `MaxMetaspaceSize=512m` (default) and Iceberg catalogs
consuming ~30-80 MB each, ~10 catalogs exhaust the limit.
This patch addresses three causes:
1. **No ClassLoader sharing** — same-type catalogs loaded identical
classes into separate Metaspace regions
2. **ClassLoader leak in `testConnection()`** — the throwaway wrapper
was never closed after a connection test
3. **Inconsistent cleanup** — only 2 of 9+ catalog types called
`ClassLoaderResourceCleanerUtils`; centralizing it in the pool ensures
every type is cleaned on final release
Fix: #10093
### Does this PR introduce _any_ user-facing change?
Yes. A new optional server configuration is added (since 2.0.0):
- `gravitino.catalog.classloader.sharing.enabled` — whether to share
ClassLoaders across catalogs with identical isolation-relevant
properties. Default is `true`. Set to `false` to give each catalog its
own dedicated ClassLoader as in previous releases.
### How was this patch tested?
**Unit tests** (`TestClassLoaderPool` — 19 tests): acquire/release
semantics, reference counting, concurrent access with 20 threads,
close-during-acquire race, double-release resilience, Kerberos key
isolation, backend URI isolation (metastore URIs, JDBC URLs, `uri`,
fs.defaultFS), authorization provider isolation, package property
isolation.
**Integration tests** (`TestClassLoaderPoolIntegration` — 6 tests):
same-type catalogs share one ClassLoader; dropping one doesn't affect
others; `testConnection` on the same key doesn't break a live catalog;
manager close cleans up the pool; with sharing disabled each catalog
gets a distinct ClassLoader; with sharing disabled dropping one doesn't
affect another.
**Docker integration tests** (`@Tag("gravitino-docker-test")`, real
MySQL): `IcebergClassLoaderPoolIT` (3 tests) verifies over a live JDBC
backend that catalogs on the same `uri` share a ClassLoader while
different `uri`s isolate, and that dropping a shared catalog / running
`testConnection` keeps a sibling usable. `TestMultipleJDBCLoad` covers
multiple Iceberg JDBC catalogs on one server. Both scope Iceberg's
metadata lookups with `nullCatalogMeansCurrent=true` (test-only) so
catalogs on separate databases of a shared MySQL container don't see
each other's `iceberg_tables`.
**Existing tests**: `TestCatalogManager` passes without modification;
`TestJdbcCatalogOperations` was trimmed (the driver-deregistration path
it covered is now handled by the pool).
**Benchmark** (JDK 17, `-XX:MaxMetaspaceSize=512m`, `fileset` provider,
10 concurrent threads):
Metaspace growth (committed KB):
| Catalogs | Baseline (`main`) | ClassLoaderPool | Reduction |
|---|---|---|---|
| 100 | +890 | +261 | 3.4x |
| 500 | +3,280 | +82 | 40x |
| 1,000 | +6,416 | +9 | 713x |
| 5,000 | +13,969 | +67 | 209x |
| 10,000 | +40,394 | +11 | **3,672x** |
Classes loaded:
| Catalogs | Baseline | ClassLoaderPool | Reduction |
|---|---|---|---|
| 1,000 | 21,772 | 11,387 | 48% |
| 10,000 | 60,373 | 11,961 | **80%** |
Baseline Metaspace grows O(N) with catalog count. The pool stays flat at
~8.7 MB — O(number of distinct keys). No OOM or performance regression
on either version. For Iceberg catalogs (~50 MB/ClassLoader), baseline
OOMs at ~10 catalogs; with the pool, catalogs sharing the same key reuse
a single ClassLoader, so Metaspace scales with the number of distinct
configurations rather than the number of catalog instances.
---
.../catalog/fileset/FilesetCatalogOperations.java | 3 -
.../catalog/hive/HiveCatalogOperations.java | 2 -
.../catalog/jdbc/JdbcCatalogOperations.java | 17 -
.../catalog/jdbc/TestJdbcCatalogOperations.java | 71 +---
.../iceberg/IcebergCatalogOperations.java | 2 -
.../integration/test/IcebergClassLoaderPoolIT.java | 265 +++++++++++++
.../integration/test/TestMultipleJDBCLoad.java | 12 +-
.../lakehouse/paimon/PaimonCatalogOperations.java | 2 -
conf/gravitino.conf.template | 4 +
.../main/java/org/apache/gravitino/Configs.java | 12 +
.../apache/gravitino/catalog/CatalogManager.java | 207 ++++++++--
.../org/apache/gravitino/utils/ClassLoaderKey.java | 70 ++++
.../apache/gravitino/utils/ClassLoaderPool.java | 216 +++++++++++
.../gravitino/utils/IsolatedClassLoader.java | 10 +
.../gravitino/utils/PooledClassLoaderEntry.java | 101 +++++
.../catalog/TestClassLoaderPoolIntegration.java | 349 +++++++++++++++++
.../gravitino/utils/TestClassLoaderPool.java | 428 +++++++++++++++++++++
docs/gravitino-server-config.md | 1 +
.../iceberg/common/ops/IcebergCatalogWrapper.java | 74 ----
.../iceberg/service/CatalogWrapperForREST.java | 5 -
.../integration/test/util/TestDatabaseName.java | 14 +
21 files changed, 1654 insertions(+), 211 deletions(-)
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 329acf7495..6f2e7b3b86 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,7 +104,6 @@ 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;
@@ -993,8 +992,6 @@ 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 99bcc1431d..7013930a18 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,7 +89,6 @@ 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;
@@ -197,7 +196,6 @@ public class HiveCatalogOperations
clientPool.close();
clientPool = null;
}
-
ClassLoaderResourceCleanerUtils.closeClassLoaderResource(this.getClass().getClassLoader());
}
/**
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
index e7d4fde6a5..94850f38a3 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
@@ -75,7 +75,6 @@ import
org.apache.gravitino.rel.expressions.distributions.Distribution;
import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
-import org.apache.gravitino.utils.IsolatedClassLoader;
import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
import org.slf4j.Logger;
@@ -201,14 +200,6 @@ public class JdbcCatalogOperations implements
CatalogOperations, SupportsSchemas
metricsSystem.unregister(catalogMetricsSource);
}
DataSourceUtils.closeDataSource(dataSource);
- try {
- Driver driver = getDriver();
- if (driver != null) {
- deregisterDriver(driver);
- }
- } catch (SQLException e) {
- LOG.warn("Failed to deregister JDBC driver", e);
- }
}
/**
@@ -622,12 +613,4 @@ public class JdbcCatalogOperations implements
CatalogOperations, SupportsSchemas
private static String currentUser() {
return PrincipalUtils.getCurrentUserName();
}
-
- public void deregisterDriver(Driver driver) throws SQLException {
- if (driver.getClass().getClassLoader().getClass()
- == IsolatedClassLoader.CUSTOM_CLASS_LOADER_CLASS) {
- DriverManager.deregisterDriver(driver);
- LOG.info("Driver {} has been deregistered...", driver);
- }
- }
}
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperations.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperations.java
index d7f711e20f..964e632db2 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperations.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperations.java
@@ -20,8 +20,6 @@ package org.apache.gravitino.catalog.jdbc;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
-import java.sql.Driver;
-import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.HashMap;
import javax.sql.DataSource;
@@ -29,8 +27,6 @@ import org.apache.commons.dbcp2.BasicDataSource;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
-import org.apache.gravitino.catalog.jdbc.converter.JdbcExceptionConverter;
-import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
import
org.apache.gravitino.catalog.jdbc.converter.SqliteColumnDefaultValueConverter;
import org.apache.gravitino.catalog.jdbc.converter.SqliteExceptionConverter;
import org.apache.gravitino.catalog.jdbc.converter.SqliteTypeConverter;
@@ -81,76 +77,15 @@ public class TestJdbcCatalogOperations {
}
@Test
- public void testCloseDeregisterDriver() throws SQLException {
- TestableJdbcCatalogOperations catalogOperations =
- new TestableJdbcCatalogOperations(
- new SqliteExceptionConverter(),
- new SqliteTypeConverter(),
- new SqliteDatabaseOperations("/illegal/path"),
- new SqliteTableOperations(),
- new SqliteColumnDefaultValueConverter());
-
catalogOperations.setDriver(DriverManager.getDriver("jdbc:sqlite::memory:"));
-
- Assertions.assertDoesNotThrow(catalogOperations::close);
- Assertions.assertTrue(catalogOperations.isDeregisterCalled());
- }
-
- @Test
- public void testCloseIgnoreGetDriverException() {
- TestableJdbcCatalogOperations catalogOperations =
- new TestableJdbcCatalogOperations(
+ public void testCloseDoesNotThrow() {
+ JdbcCatalogOperations catalogOperations =
+ new JdbcCatalogOperations(
new SqliteExceptionConverter(),
new SqliteTypeConverter(),
new SqliteDatabaseOperations("/illegal/path"),
new SqliteTableOperations(),
new SqliteColumnDefaultValueConverter());
- catalogOperations.setThrowExceptionInGetDriver(true);
Assertions.assertDoesNotThrow(catalogOperations::close);
}
-
- private static class TestableJdbcCatalogOperations extends
JdbcCatalogOperations {
- private Driver driver;
- private boolean deregisterCalled;
- private boolean throwExceptionInGetDriver;
-
- private TestableJdbcCatalogOperations(
- JdbcExceptionConverter exceptionConverter,
- JdbcTypeConverter jdbcTypeConverter,
- SqliteDatabaseOperations databaseOperation,
- SqliteTableOperations tableOperation,
- SqliteColumnDefaultValueConverter columnDefaultValueConverter) {
- super(
- exceptionConverter,
- jdbcTypeConverter,
- databaseOperation,
- tableOperation,
- columnDefaultValueConverter);
- }
-
- @Override
- protected Driver getDriver() throws SQLException {
- if (throwExceptionInGetDriver) {
- throw new SQLException("failed to get driver");
- }
- return driver;
- }
-
- @Override
- public void deregisterDriver(Driver driver) {
- this.deregisterCalled = true;
- }
-
- private void setDriver(Driver driver) {
- this.driver = driver;
- }
-
- private void setThrowExceptionInGetDriver(boolean
throwExceptionInGetDriver) {
- this.throwExceptionInGetDriver = throwExceptionInGetDriver;
- }
-
- private boolean isDeregisterCalled() {
- return deregisterCalled;
- }
- }
}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
index 3a75bad5ee..6e7346f636 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
@@ -72,7 +72,6 @@ import
org.apache.gravitino.rel.expressions.distributions.Distributions;
import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
-import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
@@ -175,7 +174,6 @@ public class IcebergCatalogOperations
if (null != icebergCatalogWrapper) {
try {
icebergCatalogWrapper.close();
-
ClassLoaderResourceCleanerUtils.closeClassLoaderResource(this.getClass().getClassLoader());
} catch (Exception e) {
LOG.warn("Failed to close Iceberg catalog", e);
}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/IcebergClassLoaderPoolIT.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/IcebergClassLoaderPoolIT.java
new file mode 100644
index 0000000000..b763087229
--- /dev/null
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/IcebergClassLoaderPoolIT.java
@@ -0,0 +1,265 @@
+/*
+ * 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.catalog.lakehouse.iceberg.integration.test;
+
+import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogPropertiesMetadata.GRAVITINO_JDBC_PASSWORD;
+import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogPropertiesMetadata.GRAVITINO_JDBC_USER;
+
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.catalog.CatalogManager;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.integration.test.container.MySQLContainer;
+import org.apache.gravitino.integration.test.util.BaseIT;
+import org.apache.gravitino.integration.test.util.ITUtils;
+import org.apache.gravitino.integration.test.util.TestDatabaseName;
+import org.apache.gravitino.utils.RandomNameUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Docker-backed integration tests for the {@code ClassLoaderPool} using a
real MySQL Iceberg JDBC
+ * backend. These complement the counting-only unit tests in {@code
TestClassLoaderPoolIntegration}
+ * by exercising the two runtime properties the pool actually claims:
+ *
+ * <ul>
+ * <li><b>Key isolation (the {@code uri} blind spot):</b> Iceberg's JDBC
backend uses the {@code
+ * uri} catalog property, not {@code jdbc-url}. Two MySQL-backed Iceberg
catalogs on different
+ * databases must NOT share a ClassLoader (otherwise they would
cross-contaminate the
+ * per-ClassLoader {@code DriverManager} registry); two catalogs on the
same {@code uri} must
+ * share one.
+ * <li><b>Lifecycle / no premature cleanup:</b> dropping one catalog that
shares a ClassLoader
+ * must not deregister the JDBC driver (or shut down MySQL's {@code
+ * AbandonedConnectionCleanupThread}) while a sibling catalog is still
live, and {@code
+ * testConnection} must acquire+release the shared key without breaking
a live catalog.
+ * </ul>
+ *
+ * <p>ClassLoader-identity assertions require reflecting into the in-process
server, so they run
+ * only in embedded mode; the end-to-end "still works after drop /
testConnection" assertions run in
+ * any mode via the client.
+ */
+@Tag("gravitino-docker-test")
+public class IcebergClassLoaderPoolIT extends BaseIT {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergClassLoaderPoolIT.class);
+
+ private static final TestDatabaseName DB_A =
+ TestDatabaseName.MYSQL_TEST_ICEBERG_CLASSLOADER_POOL_A;
+ private static final TestDatabaseName DB_B =
+ TestDatabaseName.MYSQL_TEST_ICEBERG_CLASSLOADER_POOL_B;
+ private static final String PROVIDER = "lakehouse-iceberg";
+
+ private static MySQLContainer mySQLContainer;
+
+ // Metalakes created by tests, dropped (force) in tearDown so no state leaks
across tests/runs
+ // (important in deploy mode where the entity store persists between runs).
+ private final List<String> createdMetalakes = new ArrayList<>();
+
+ @BeforeAll
+ public void startup() throws IOException {
+ containerSuite.startMySQLContainer(DB_A);
+ mySQLContainer = containerSuite.getMySQLContainer();
+ // Second database on the same MySQL container to produce a distinct
backend URI.
+ mySQLContainer.createDatabase(DB_B);
+ }
+
+ @AfterEach
+ public void cleanup() {
+ for (String metalakeName : createdMetalakes) {
+ try {
+ client.dropMetalake(metalakeName, true);
+ } catch (Exception e) {
+ LOG.warn("Failed to drop metalake {} during cleanup", metalakeName, e);
+ }
+ }
+ createdMetalakes.clear();
+ }
+
+ private GravitinoMetalake createMetalake(String name) {
+ GravitinoMetalake metalake = client.createMetalake(name, "comment",
Collections.emptyMap());
+ createdMetalakes.add(name);
+ return metalake;
+ }
+
+ private Map<String, String> icebergMysqlConf(TestDatabaseName db) throws
Exception {
+ Map<String, String> conf = Maps.newHashMap();
+ // Scope JDBC metadata lookups to the connected database via
nullCatalogMeansCurrent=true.
+ // These tests deliberately place DB_A and DB_B on the SAME MySQL server
(to produce two
+ // distinct backend URIs for the ClassLoader isolation assertion).
Iceberg's JdbcCatalog
+ // checks whether its `iceberg_tables` control table already exists with
+ // getTables(catalog=null, schema=null, "iceberg_tables", null) before
creating it. Under
+ // Connector/J's default (nullCatalogMeansCurrent=false) that call scans
EVERY database on the
+ // server, so a catalog on DB_B would "see" DB_A's iceberg_tables, skip
creating its own, and
+ // then fail real queries with "table doesn't exist". Setting
nullCatalogMeansCurrent=true
+ // scopes the check to the current database. (This mirrors the recommended
setting for running
+ // multiple Iceberg JDBC catalogs against separate databases of one MySQL
server.)
+ conf.put(
+ IcebergConfig.CATALOG_URI.getKey(),
+ mySQLContainer.getJdbcUrl(db) + "?nullCatalogMeansCurrent=true");
+ conf.put(IcebergConfig.CATALOG_BACKEND.getKey(), "jdbc");
+ // Unique backend name per catalog so each catalog's JDBC namespace
registry
+ // (Iceberg's iceberg_namespace_properties, keyed by the initialize()
name) is isolated even
+ // when catalogs share a database. Without this, all catalogs default to
backend name "jdbc"
+ // and would see each other's namespaces, causing cross-test contamination.
+ // `catalog-backend-name`
+ // is NOT in DEFAULT_ISOLATION_PROPERTY_KEYS, so catalogs on the same
`uri` still share a
+ // ClassLoader (which is what the sharing/lifecycle assertions rely on).
+ conf.put(
+ IcebergConfig.CATALOG_BACKEND_NAME.getKey(),
RandomNameUtils.genRandomName("clp_backend"));
+ // Unique warehouse dir per catalog to avoid stale /tmp state leaking
across catalogs and runs.
+ conf.put(
+ IcebergConfig.CATALOG_WAREHOUSE.getKey(),
+ "file:///tmp/" + RandomNameUtils.genRandomName("iceberg_clp_wh"));
+ conf.put(IcebergConfig.JDBC_DRIVER.getKey(),
mySQLContainer.getDriverClassName(db));
+ conf.put(GRAVITINO_JDBC_USER, mySQLContainer.getUsername());
+ conf.put(GRAVITINO_JDBC_PASSWORD, mySQLContainer.getPassword());
+ return conf;
+ }
+
+ private boolean isEmbedded() {
+ return ITUtils.EMBEDDED_TEST_MODE.equals(testMode);
+ }
+
+ private Object classLoaderOf(String metalake, String catalog) throws
IllegalAccessException {
+ CatalogManager catalogManager =
GravitinoEnv.getInstance().catalogManager();
+ // Use loadCatalogAndWrap (rather than the Caffeine cache directly) so
this test module does not
+ // need caffeine on its compile classpath. The catalog was already loaded
server-side above, so
+ // this returns the cached wrapper instance.
+ CatalogManager.CatalogWrapper wrapper =
+ catalogManager.loadCatalogAndWrap(NameIdentifier.of(metalake,
catalog));
+ Assertions.assertNotNull(wrapper, "wrapper should be cached for " +
metalake + "." + catalog);
+ return FieldUtils.readField(wrapper, "classLoader", true);
+ }
+
+ @Test
+ public void testSameUriShareAndDifferentUriIsolateClassLoader() throws
Exception {
+ Assumptions.assumeTrue(
+ isEmbedded(),
+ "ClassLoader-identity assertions require the in-process server
(embedded mode)");
+
+ String metalakeName = RandomNameUtils.genRandomName("clp_metalake");
+ GravitinoMetalake metalake = createMetalake(metalakeName);
+
+ // Two catalogs on the SAME uri (DB_A) must share one ClassLoader.
+ String sameA1 = RandomNameUtils.genRandomName("clp_same_a1");
+ String sameA2 = RandomNameUtils.genRandomName("clp_same_a2");
+ metalake.createCatalog(
+ sameA1, Catalog.Type.RELATIONAL, PROVIDER, "comment",
icebergMysqlConf(DB_A));
+ metalake.createCatalog(
+ sameA2, Catalog.Type.RELATIONAL, PROVIDER, "comment",
icebergMysqlConf(DB_A));
+
+ // Force server-side load so both wrappers are cached.
+ metalake.loadCatalog(sameA1);
+ metalake.loadCatalog(sameA2);
+
+ Assertions.assertSame(
+ classLoaderOf(metalakeName, sameA1),
+ classLoaderOf(metalakeName, sameA2),
+ "Iceberg catalogs sharing the same backend uri must share a
ClassLoader");
+
+ // A catalog on a DIFFERENT uri (DB_B) must NOT share the ClassLoader —
this is the `uri`
+ // isolation-key fix; without `uri` in DEFAULT_ISOLATION_PROPERTY_KEYS
these would collide.
+ String otherB = RandomNameUtils.genRandomName("clp_other_b");
+ metalake.createCatalog(
+ otherB, Catalog.Type.RELATIONAL, PROVIDER, "comment",
icebergMysqlConf(DB_B));
+ metalake.loadCatalog(otherB);
+
+ Assertions.assertNotSame(
+ classLoaderOf(metalakeName, sameA1),
+ classLoaderOf(metalakeName, otherB),
+ "Iceberg catalogs with different backend uris must NOT share a
ClassLoader");
+ }
+
+ @Test
+ public void testDropOneSharedCatalogKeepsSiblingUsable() throws Exception {
+ String metalakeName = RandomNameUtils.genRandomName("clp_metalake");
+ GravitinoMetalake metalake = createMetalake(metalakeName);
+
+ String catalog1 = RandomNameUtils.genRandomName("clp_drop_1");
+ String catalog2 = RandomNameUtils.genRandomName("clp_drop_2");
+ metalake.createCatalog(
+ catalog1, Catalog.Type.RELATIONAL, PROVIDER, "comment",
icebergMysqlConf(DB_A));
+ Catalog cat2 =
+ metalake.createCatalog(
+ catalog2, Catalog.Type.RELATIONAL, PROVIDER, "comment",
icebergMysqlConf(DB_A));
+
+ // Exercise catalog2's real JDBC backend once so its driver is
registered/used.
+ Assertions.assertEquals(0, cat2.asSchemas().listSchemas().length);
+
+ // Drop catalog1 — since it shares the pooled ClassLoader (and the MySQL
driver /
+ // AbandonedConnectionCleanupThread) with catalog2, the final-release
cleanup must NOT run.
+ metalake.disableCatalog(catalog1);
+ Assertions.assertTrue(metalake.dropCatalog(catalog1, true));
+
+ // Issue a REAL query against catalog2 after catalog1's drop. If the
shared driver had been
+ // deregistered / the cleanup thread shut down, this JDBC call would fail.
+ Catalog reloaded = metalake.loadCatalog(catalog2);
+ String schemaName = RandomNameUtils.genRandomName("clp_schema");
+ Assertions.assertDoesNotThrow(
+ () -> reloaded.asSchemas().createSchema(schemaName, null,
Collections.emptyMap()));
+ Assertions.assertTrue(reloaded.asSchemas().schemaExists(schemaName));
+ }
+
+ @Test
+ public void testTestConnectionKeepsLiveSharedCatalogUsable() throws
Exception {
+ String metalakeName = RandomNameUtils.genRandomName("clp_metalake");
+ GravitinoMetalake metalake = createMetalake(metalakeName);
+
+ String liveCatalog = RandomNameUtils.genRandomName("clp_live");
+ Catalog live =
+ metalake.createCatalog(
+ liveCatalog, Catalog.Type.RELATIONAL, PROVIDER, "comment",
icebergMysqlConf(DB_A));
+ Assertions.assertEquals(0, live.asSchemas().listSchemas().length);
+
+ // testConnection builds a throwaway wrapper that acquires + releases the
SAME pooled key.
+ // The release must only decrement the refCount, not run the destructive
final cleanup, because
+ // the live catalog still shares the ClassLoader / driver.
+ String probeCatalog = RandomNameUtils.genRandomName("clp_probe");
+ Assertions.assertDoesNotThrow(
+ () ->
+ metalake.testConnection(
+ probeCatalog,
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "comment",
+ icebergMysqlConf(DB_A)));
+
+ // The live catalog must still perform real JDBC work after the probe's
acquire/release cycle.
+ Catalog reloaded = metalake.loadCatalog(liveCatalog);
+ String schemaName = RandomNameUtils.genRandomName("clp_schema");
+ Assertions.assertDoesNotThrow(
+ () -> reloaded.asSchemas().createSchema(schemaName, null,
Collections.emptyMap()));
+ Assertions.assertTrue(reloaded.asSchemas().schemaExists(schemaName));
+ }
+}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/TestMultipleJDBCLoad.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/TestMultipleJDBCLoad.java
index c2e5077c7b..ca9763efdb 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/TestMultipleJDBCLoad.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/TestMultipleJDBCLoad.java
@@ -76,7 +76,14 @@ public class TestMultipleJDBCLoad extends BaseIT {
+ "?user="
+ mySQLContainer.getUsername()
+ "&password="
- + mySQLContainer.getPassword();
+ + mySQLContainer.getPassword()
+ // Scope Iceberg's iceberg_tables existence check to the current
database. Other tests
+ // in this module (e.g. IcebergClassLoaderPoolIT) create Iceberg
JDBC catalogs in other
+ // databases of the SAME shared MySQL container. Under
Connector/J's default
+ // (nullCatalogMeansCurrent=false), Iceberg's getTables(null,
null, "iceberg_tables",
+ // null) check would see those other databases' control tables and
skip creating this
+ // catalog's own, breaking later queries. See
IcebergClassLoaderPoolIT for details.
+ + "&nullCatalogMeansCurrent=true";
icebergMysqlConf.put(IcebergConfig.CATALOG_URI.getKey(), jdbcUrl);
icebergMysqlConf.put(IcebergConfig.CATALOG_BACKEND.getKey(), "jdbc");
@@ -121,7 +128,8 @@ public class TestMultipleJDBCLoad extends BaseIT {
Map<String, String> icebergMysqlConf = Maps.newHashMap();
icebergMysqlConf.put(
- IcebergConfig.CATALOG_URI.getKey(),
mySQLContainer.getJdbcUrl(TEST_DB_NAME));
+ IcebergConfig.CATALOG_URI.getKey(),
+ mySQLContainer.getJdbcUrl(TEST_DB_NAME) +
"?nullCatalogMeansCurrent=true");
icebergMysqlConf.put(IcebergConfig.CATALOG_BACKEND.getKey(), "jdbc");
icebergMysqlConf.put(IcebergConfig.CATALOG_WAREHOUSE.getKey(),
"file:///tmp/iceberg-jdbc");
icebergMysqlConf.put(
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
index e3cc127615..bcfc07d0c9 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
@@ -70,7 +70,6 @@ import
org.apache.gravitino.rel.expressions.distributions.Strategy;
import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
-import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils;
import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.paimon.catalog.Catalog;
@@ -559,7 +558,6 @@ public class PaimonCatalogOperations
if (paimonCatalogOps != null) {
try {
paimonCatalogOps.close();
-
ClassLoaderResourceCleanerUtils.closeClassLoaderResource(this.getClass().getClassLoader());
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git a/conf/gravitino.conf.template b/conf/gravitino.conf.template
index 8336246d6a..4517c07137 100644
--- a/conf/gravitino.conf.template
+++ b/conf/gravitino.conf.template
@@ -60,6 +60,10 @@ gravitino.entity.store.relational.jdbcPassword = gravitino
# THE CONFIGURATION FOR Gravitino CATALOG
# The interval in milliseconds to evict the catalog cache
gravitino.catalog.cache.evictionIntervalMs = 3600000
+# Whether to share a ClassLoader across catalogs that have identical
isolation-relevant properties.
+# When true (default), such catalogs reuse one ClassLoader, reducing Metaspace
usage. Set to false
+# to give every catalog its own ClassLoader, as in releases before 2.0.0.
+gravitino.catalog.classloader.sharing.enabled = true
# THE CONFIGURATION FOR Gravitino Entity Cache
# Whether to enable the cached Entity store.
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index 134f36f2e5..08702ae734 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -220,6 +220,18 @@ public class Configs {
.booleanConf()
.createWithDefault(true);
+ public static final ConfigEntry<Boolean> CATALOG_CLASSLOADER_SHARING_ENABLED
=
+ new ConfigBuilder("gravitino.catalog.classloader.sharing.enabled")
+ .doc(
+ "Whether to share ClassLoaders across catalogs with identical
isolation-relevant "
+ + "properties. When true (default), catalogs with the same
isolation key reuse "
+ + "a single ClassLoader, significantly reducing Metaspace
memory usage. When "
+ + "false, each catalog gets its own dedicated ClassLoader as
in previous "
+ + "releases.")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .booleanConf()
+ .createWithDefault(true);
+
public static final ConfigEntry<String> AUTHENTICATOR =
new ConfigBuilder("gravitino.authenticator")
.doc(
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 75a8c0f72a..35d0c3fdb5 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -108,8 +108,11 @@ import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.relational.SupportsEntityChangeLog;
+import org.apache.gravitino.utils.ClassLoaderKey;
+import org.apache.gravitino.utils.ClassLoaderPool;
import org.apache.gravitino.utils.IsolatedClassLoader;
import org.apache.gravitino.utils.NamespaceUtil;
+import org.apache.gravitino.utils.PooledClassLoaderEntry;
import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.gravitino.utils.ThrowableFunction;
import org.slf4j.Logger;
@@ -125,15 +128,53 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
private static final Set<String> CONTRIB_CATALOGS_TYPES =
ImmutableSet.of("jdbc-oceanbase", "jdbc-clickhouse", "jdbc-hologres");
+ // Isolation property keys included in the ClassLoaderKey. They cover the
catalog property
+ // dimensions that determine the classpath or anchor per-ClassLoader static
state.
+ //
+ // Classpath dimensions:
+ // - package: determines which JARs are loaded
+ // - authorization-provider: determines which authorization plugin JARs
are loaded
+ // Static-state dimensions:
+ // - authentication.type/kerberos.principal/kerberos.keytab-uri: Hadoop
UGI is per-ClassLoader
+ // - metastore.uris: HiveConf static configuration space
+ // - jdbc-url: JDBC DriverManager global registry per ClassLoader (JDBC
catalogs)
+ // - uri: backend URI for Iceberg/Paimon/Hudi catalogs. Their JDBC
backends register drivers
+ // under this key (not "jdbc-url") in the per-ClassLoader DriverManager
registry, so it must
+ // be an isolation dimension — otherwise two MySQL-backed Iceberg
catalogs on different
+ // databases would share one ClassLoader and cross-contaminate the
driver registry.
+ // - fs.defaultFS: Hadoop FileSystem.CACHE per ClassLoader
+ static final Set<String> DEFAULT_ISOLATION_PROPERTY_KEYS =
+ ImmutableSet.of(
+ Catalog.PROPERTY_PACKAGE,
+ Catalog.AUTHORIZATION_PROVIDER,
+ "authentication.type",
+ "authentication.kerberos.principal",
+ "authentication.kerberos.keytab-uri",
+ "metastore.uris",
+ "jdbc-url",
+ "uri",
+ "fs.defaultFS");
+
/** Wrapper class for a catalog instance and its class loader. */
public static class CatalogWrapper {
private BaseCatalog catalog;
private IsolatedClassLoader classLoader;
+ private ClassLoaderPool pool;
+ private PooledClassLoaderEntry poolEntry;
+ private boolean closed = false;
+
+ /** Non-pooled constructor: each catalog owns its ClassLoader exclusively.
*/
+ CatalogWrapper(IsolatedClassLoader classLoader) {
+ this.classLoader = classLoader;
+ }
- public CatalogWrapper(BaseCatalog catalog, IsolatedClassLoader
classLoader) {
- this.catalog = catalog;
+ /** Pooled constructor: ClassLoader is managed by the pool with reference
counting. */
+ CatalogWrapper(
+ IsolatedClassLoader classLoader, ClassLoaderPool pool,
PooledClassLoaderEntry poolEntry) {
this.classLoader = classLoader;
+ this.pool = pool;
+ this.poolEntry = poolEntry;
}
public BaseCatalog catalog() {
@@ -243,7 +284,13 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
return classLoader.withClassLoader(cl -> catalog.capability());
}
- public void close() {
+ public synchronized void close() {
+ if (closed) {
+ // Idempotent: a second close() must not re-run pool release or
classloader cleanup.
+ return;
+ }
+ closed = true;
+
try {
classLoader.withClassLoader(
cl -> {
@@ -255,9 +302,18 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
});
} catch (Exception e) {
LOG.warn("Failed to close catalog", e);
+ } finally {
+ // Release the pool reference (or clean up the dedicated ClassLoader)
in a finally so a
+ // failure while closing the catalog cannot permanently leak the
pooled ClassLoader
+ // reference (close() is idempotent, so a retry would otherwise skip
this).
+ if (poolEntry != null) {
+ pool.release(poolEntry);
+ poolEntry = null;
+ } else if (pool == null) {
+ // Non-pooled path (e.g., sharing disabled or
CATALOG_LOAD_ISOLATED=false)
+ ClassLoaderPool.cleanupClassLoader(classLoader);
+ }
}
-
- classLoader.close();
}
private SupportsSchemas asSchemas() {
@@ -291,6 +347,10 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
private final Config config;
+ private final ClassLoaderPool classLoaderPool = new ClassLoaderPool();
+
+ private final boolean classLoaderSharingEnabled;
+
@Getter private final Cache<NameIdentifier, CatalogWrapper> catalogCache;
private final EntityStore store;
@@ -318,6 +378,7 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
this.config = config;
this.store = store;
this.idGenerator = idGenerator;
+ this.classLoaderSharingEnabled =
config.get(Configs.CATALOG_CLASSLOADER_SHARING_ENABLED);
long cacheEvictionIntervalInMs =
config.get(Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS);
this.catalogCache =
@@ -371,6 +432,7 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
localMutationCounts.clear();
}
catalogCache.invalidateAll();
+ classLoaderPool.close();
}
/**
@@ -647,11 +709,15 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
.build();
CatalogWrapper wrapper = createCatalogWrapper(dummyEntity, mergedConfig);
- wrapper.doWithCatalogOps(
- c -> {
- c.testConnection(ident, type, provider, comment, mergedConfig);
- return null;
- });
+ try {
+ wrapper.doWithCatalogOps(
+ c -> {
+ c.testConnection(ident, type, provider, comment, mergedConfig);
+ return null;
+ });
+ } finally {
+ wrapper.close();
+ }
} catch (GravitinoRuntimeException e) {
throw e;
} catch (Exception e) {
@@ -1123,34 +1189,85 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
Map<String, String> conf = entity.getProperties();
String provider = entity.getProvider();
+ if (!classLoaderSharingEnabled) {
+ return createNonPooledCatalogWrapper(provider, conf, entity,
propsToValidate);
+ }
+
+ ClassLoaderKey key = buildClassLoaderKey(provider, conf);
+ PooledClassLoaderEntry poolEntry =
+ classLoaderPool.acquire(key, () -> createClassLoader(provider, conf));
+ try {
+ CatalogWrapper wrapper =
+ initCatalogWrapper(
+ new CatalogWrapper(poolEntry.classLoader(), classLoaderPool,
poolEntry),
+ entity,
+ propsToValidate);
+ return wrapper;
+ } catch (Exception e) {
+ classLoaderPool.release(poolEntry);
+ throw e;
+ }
+ }
+
+ private CatalogWrapper createNonPooledCatalogWrapper(
+ String provider,
+ Map<String, String> conf,
+ CatalogEntity entity,
+ @Nullable Map<String, String> propsToValidate) {
IsolatedClassLoader classLoader = createClassLoader(provider, conf);
- BaseCatalog<?> catalog = createBaseCatalog(classLoader, entity);
-
- CatalogWrapper wrapper = new CatalogWrapper(catalog, classLoader);
- // Validate catalog properties and initialize the config
- classLoader.withClassLoader(
- cl -> {
- validatePropertyForCreate(catalog.catalogPropertiesMetadata(),
propsToValidate);
-
- // Call wrapper.catalog.properties() to make BaseCatalog#properties
in IsolatedClassLoader
- // not null. Why do we do this? Because wrapper.catalog.properties()
needs to be called in
- // the IsolatedClassLoader, as it needs to load the specific catalog
class
- // such as HiveCatalog or similar. To simplify, we will preload the
value of properties
- // so that AppClassLoader can get the value of properties.
- wrapper.catalog.properties();
- wrapper.catalog.capability();
-
- // Eagerly initialize opted-in catalogs so a misconfiguration fails
at create rather than
- // on first use. The backend translates failures into the
caller-error vs
- // dependency-unavailable taxonomy; both types are forwarded by the
passthrough below.
- if (propsToValidate != null &&
wrapper.catalog.shouldValidateOnCreate()) {
- wrapper.catalog.ops();
- }
- return null;
- },
- IllegalArgumentException.class,
- ConnectionFailedException.class);
+ try {
+ return initCatalogWrapper(new CatalogWrapper(classLoader), entity,
propsToValidate);
+ } catch (Exception e) {
+ ClassLoaderPool.cleanupClassLoader(classLoader);
+ throw e;
+ }
+ }
+ /**
+ * Creates the catalog instance, validates properties, and preloads
property/capability values
+ * into the given wrapper.
+ */
+ private CatalogWrapper initCatalogWrapper(
+ CatalogWrapper wrapper, CatalogEntity entity, @Nullable Map<String,
String> propsToValidate) {
+ BaseCatalog<?> catalog = createBaseCatalog(wrapper.classLoader, entity);
+ wrapper.catalog = catalog;
+ try {
+ wrapper.classLoader.withClassLoader(
+ cl -> {
+ validatePropertyForCreate(catalog.catalogPropertiesMetadata(),
propsToValidate);
+ // Preload properties() and capability() inside the
IsolatedClassLoader so that
+ // AppClassLoader can read them later without needing the isolated
context.
+ catalog.properties();
+ catalog.capability();
+
+ // Eagerly initialize opted-in catalogs so a misconfiguration
fails at create rather
+ // than on first use. The backend translates failures into the
caller-error vs
+ // dependency-unavailable taxonomy; both types are forwarded by
the passthrough below.
+ if (propsToValidate != null && catalog.shouldValidateOnCreate()) {
+ catalog.ops();
+ }
+ return null;
+ },
+ IllegalArgumentException.class,
+ ConnectionFailedException.class);
+ } catch (RuntimeException e) {
+ // Close the partially-initialized catalog (which releases its
authorizationPlugin and other
+ // resources) before the caller tears down or releases the ClassLoader.
Otherwise a creation
+ // failure — e.g. invalid properties on a catalog that has an
authorization-provider — leaks
+ // the plugin instance, since the caller's catch only releases the
ClassLoader, not the
+ // catalog.
+ try {
+ wrapper.classLoader.withClassLoader(
+ cl -> {
+ catalog.close();
+ return null;
+ });
+ } catch (Exception closeEx) {
+ LOG.warn("Failed to close catalog after initialization failure",
closeEx);
+ }
+ wrapper.catalog = null;
+ throw e;
+ }
return wrapper;
}
@@ -1162,6 +1279,11 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
* @return The resolved properties.
*/
private Map<String, String> getResolvedProperties(CatalogEntity entity) {
+ // Resolve properties through the cached wrapper (loadCatalogAndWrap),
which reuses the
+ // pooled/dedicated ClassLoader and the CatalogWrapper cache. This avoids
building and tearing
+ // down a throwaway BaseCatalog (and leaking its authorizationPlugin) on
every listCatalogsInfo
+ // call, and keeps the classLoaderSharingEnabled branching in a single
place
+ // (createCatalogWrapper).
CatalogWrapper catalogWrapper =
loadCatalogAndWrap(entity.nameIdentifier());
return catalogWrapper.classLoader.withClassLoader(
cl -> catalogWrapper.catalog.properties(), RuntimeException.class);
@@ -1175,6 +1297,19 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
return catalog;
}
+ private ClassLoaderKey buildClassLoaderKey(String provider, Map<String,
String> conf) {
+ Map<String, String> isolationProps = new HashMap<>();
+ if (conf != null) {
+ for (String key : DEFAULT_ISOLATION_PROPERTY_KEYS) {
+ String value = conf.get(key);
+ if (value != null) {
+ isolationProps.put(key, value);
+ }
+ }
+ }
+ return new ClassLoaderKey(provider, isolationProps.isEmpty() ? null :
isolationProps);
+ }
+
private IsolatedClassLoader createClassLoader(String provider, Map<String,
String> conf) {
if (config.get(Configs.CATALOG_LOAD_ISOLATED)) {
String catalogPkgPath = buildPkgPath(conf, provider);
diff --git a/core/src/main/java/org/apache/gravitino/utils/ClassLoaderKey.java
b/core/src/main/java/org/apache/gravitino/utils/ClassLoaderKey.java
new file mode 100644
index 0000000000..fbc5677c8e
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/utils/ClassLoaderKey.java
@@ -0,0 +1,70 @@
+/*
+ * 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.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+
+/**
+ * Key used to identify a shared ClassLoader in the {@link ClassLoaderPool}.
Stores isolation
+ * properties as a generic {@code Map<String, String>}, decoupled from any
specific property names.
+ * This makes the pool infrastructure key-agnostic — only the logic that
builds the key (in {@code
+ * CatalogManager}) needs to know which properties matter.
+ */
+public class ClassLoaderKey {
+
+ private final String provider;
+ private final Map<String, String> properties;
+
+ /**
+ * Constructs a ClassLoaderKey.
+ *
+ * @param provider The catalog provider name (e.g., "hive",
"lakehouse-iceberg").
+ * @param properties The isolation-relevant properties extracted from the
catalog configuration,
+ * or null if none.
+ */
+ public ClassLoaderKey(String provider, @Nullable Map<String, String>
properties) {
+ this.provider = Objects.requireNonNull(provider, "provider must not be
null");
+ this.properties =
+ properties == null
+ ? Collections.emptyMap()
+ : Collections.unmodifiableMap(new TreeMap<>(properties));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof ClassLoaderKey)) return false;
+ ClassLoaderKey that = (ClassLoaderKey) o;
+ return Objects.equals(provider, that.provider) &&
Objects.equals(properties, that.properties);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(provider, properties);
+ }
+
+ @Override
+ public String toString() {
+ return "ClassLoaderKey{" + "provider='" + provider + '\'' + ",
properties=" + properties + '}';
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/utils/ClassLoaderPool.java
b/core/src/main/java/org/apache/gravitino/utils/ClassLoaderPool.java
new file mode 100644
index 0000000000..fd376fb0e1
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/utils/ClassLoaderPool.java
@@ -0,0 +1,216 @@
+/*
+ * 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.io.Closeable;
+import java.net.URLClassLoader;
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.util.Enumeration;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A pool that manages shared {@link IsolatedClassLoader} instances across
catalogs with identical
+ * isolation-relevant properties (package, authorization provider, Kerberos
identity, metastore
+ * URIs, JDBC URL, default filesystem). Sharing ClassLoaders across
same-configuration catalogs
+ * significantly reduces Metaspace memory usage.
+ *
+ * <p>Thread safety is guaranteed through {@link ConcurrentHashMap#compute}
for all acquire/release
+ * operations.
+ *
+ * <p><b>Concurrency note:</b> the classloader factory (JAR scanning in {@code
acquire}) and {@link
+ * #doFinalCleanup} (thread interruption and reflective ThreadLocal cleanup in
{@code release}) run
+ * inside {@code compute()}, which holds the per-key bin lock for the
duration. This serializes
+ * create/drop of catalogs that map to the <em>same</em> key. Operations on
different keys hash to
+ * different bins and proceed concurrently, so for the expected pool size (a
handful of distinct
+ * configurations) the added latency is acceptable and keeps
acquire/release/cleanup atomic.
+ */
+public class ClassLoaderPool implements Closeable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ClassLoaderPool.class);
+
+ private final ConcurrentHashMap<ClassLoaderKey, PooledClassLoaderEntry> pool
=
+ new ConcurrentHashMap<>();
+
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ /**
+ * Acquires a ClassLoader entry for the given key. If an entry already
exists, increments the
+ * reference count. Otherwise, creates a new entry using the provided
factory.
+ *
+ * @param key The key identifying the ClassLoader configuration.
+ * @param factory A supplier that creates a new IsolatedClassLoader when
needed.
+ * @return The pooled ClassLoader entry.
+ * @throws IllegalStateException if the pool has been closed.
+ */
+ public PooledClassLoaderEntry acquire(ClassLoaderKey key,
Supplier<IsolatedClassLoader> factory) {
+ return pool.compute(
+ key,
+ (k, existing) -> {
+ if (closed.get()) {
+ throw new IllegalStateException("ClassLoaderPool is already
closed");
+ }
+ if (existing != null) {
+ existing.incrementRefCount();
+ LOG.debug("Reusing ClassLoader for key {}, refCount={}.", key,
existing.refCount());
+ return existing;
+ }
+ // If the factory throws (e.g., invalid classpath), the exception
propagates to the
+ // caller and ConcurrentHashMap leaves the key unmapped.
+ IsolatedClassLoader classLoader = factory.get();
+ PooledClassLoaderEntry newEntry = new PooledClassLoaderEntry(k,
classLoader);
+ LOG.info("Created new ClassLoader for key {}, refCount=1.", key);
+ return newEntry;
+ });
+ }
+
+ /**
+ * Releases a ClassLoader entry. Decrements the reference count and, if it
reaches zero, performs
+ * final cleanup and removes the entry from the pool.
+ *
+ * @param entry The pooled ClassLoader entry to release.
+ */
+ public void release(PooledClassLoaderEntry entry) {
+ pool.compute(
+ entry.key(),
+ (k, existing) -> {
+ if (existing == null) {
+ LOG.warn("Attempted to release a ClassLoader entry that is not in
the pool: {}", k);
+ return null;
+ }
+ int newCount = existing.decrementRefCount();
+ LOG.debug("Released ClassLoader for key {}, refCount={}.", k,
newCount);
+ if (newCount <= 0) {
+ doFinalCleanup(existing);
+ return null; // Remove from map
+ }
+ return existing;
+ });
+ }
+
+ /**
+ * Returns the current number of distinct ClassLoader entries in the pool.
+ *
+ * @return The pool size.
+ */
+ public int size() {
+ return pool.size();
+ }
+
+ /**
+ * Closes the pool and cleans up all ClassLoader entries. After this method
returns, {@link
+ * #acquire} will throw {@link IllegalStateException}. Uses per-key {@code
compute()} to ensure
+ * atomic removal and cleanup, preventing races with concurrent release
operations.
+ */
+ @Override
+ public void close() {
+ closed.set(true);
+ // Drain with a loop to catch entries inserted by concurrent acquire()
calls that were
+ // already past the closed check when we set the flag. Since closed=true
prevents any new
+ // entries from being created, this loop is guaranteed to terminate.
+ while (!pool.isEmpty()) {
+ pool.keySet().forEach(this::removeAndCleanup);
+ }
+ }
+
+ private void removeAndCleanup(ClassLoaderKey key) {
+ pool.compute(
+ key,
+ (k, existing) -> {
+ if (existing != null) {
+ if (existing.refCount() > 0) {
+ LOG.warn(
+ "Force-closing ClassLoader for key {} with {} active
reference(s)"
+ + " (pool shutting down).",
+ k,
+ existing.refCount());
+ } else {
+ LOG.info("Closing pooled ClassLoader for key {}.", k);
+ }
+ doFinalCleanup(existing);
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Performs final cleanup when a ClassLoader's reference count reaches zero.
This includes:
+ *
+ * <ol>
+ * <li>Deregistering all JDBC drivers loaded by the ClassLoader
+ * <li>Cleaning up ClassLoader resources (ThreadLocals, Hadoop FileSystem,
etc.)
+ * <li>Closing the ClassLoader itself
+ * </ol>
+ */
+ private void doFinalCleanup(PooledClassLoaderEntry entry) {
+ if (!entry.markCleanedUp()) {
+ LOG.debug("ClassLoader for key {} already cleaned up, skipping.",
entry.key());
+ return;
+ }
+
+ cleanupClassLoader(entry.classLoader());
+ LOG.info("ClassLoader for key {} has been fully cleaned up.", entry.key());
+ }
+
+ /**
+ * Performs full resource cleanup for an {@link IsolatedClassLoader}:
deregisters JDBC drivers,
+ * cleans up ClassLoader-scoped resources (ThreadLocals, Hadoop FileSystem,
etc.), and closes the
+ * ClassLoader.
+ *
+ * @param classLoader The IsolatedClassLoader to clean up.
+ */
+ public static void cleanupClassLoader(IsolatedClassLoader classLoader) {
+ try {
+ URLClassLoader internalCl = classLoader.getInternalClassLoader();
+ if (internalCl != null) {
+ deregisterAllDrivers(internalCl);
+ ClassLoaderResourceCleanerUtils.closeClassLoaderResource(internalCl);
+ }
+ } catch (Exception e) {
+ LOG.warn("Error during ClassLoader resource cleanup", e);
+ }
+ classLoader.close();
+ }
+
+ /**
+ * Deregisters all JDBC drivers that were loaded by the given ClassLoader.
+ *
+ * @param classLoader The ClassLoader whose drivers should be deregistered.
+ */
+ private static void deregisterAllDrivers(ClassLoader classLoader) {
+ // DriverManager.getDrivers() returns a snapshot in JDK 9+, so iterating
while
+ // calling deregisterDriver() is safe.
+ Enumeration<Driver> drivers = DriverManager.getDrivers();
+ while (drivers.hasMoreElements()) {
+ Driver driver = drivers.nextElement();
+ if (driver.getClass().getClassLoader() == classLoader) {
+ try {
+ DriverManager.deregisterDriver(driver);
+ LOG.info("Deregistered JDBC driver {} for ClassLoader.", driver);
+ } catch (Exception e) {
+ LOG.warn("Failed to deregister JDBC driver {}", driver, e);
+ }
+ }
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/utils/IsolatedClassLoader.java
b/core/src/main/java/org/apache/gravitino/utils/IsolatedClassLoader.java
index 82eb7213e7..ef83295720 100644
--- a/core/src/main/java/org/apache/gravitino/utils/IsolatedClassLoader.java
+++ b/core/src/main/java/org/apache/gravitino/utils/IsolatedClassLoader.java
@@ -171,6 +171,16 @@ public class IsolatedClassLoader implements Closeable {
classPathContents, Collections.emptyList(), Collections.emptyList());
}
+ /**
+ * Returns the internal URLClassLoader used by this IsolatedClassLoader.
This is used by {@link
+ * ClassLoaderPool} for resource cleanup (e.g., JDBC driver deregistration,
ThreadLocal cleanup).
+ *
+ * @return The internal URLClassLoader, or null if not yet initialized.
+ */
+ public URLClassLoader getInternalClassLoader() {
+ return classLoader;
+ }
+
/** Closes the class loader. */
@Override
public void close() {
diff --git
a/core/src/main/java/org/apache/gravitino/utils/PooledClassLoaderEntry.java
b/core/src/main/java/org/apache/gravitino/utils/PooledClassLoaderEntry.java
new file mode 100644
index 0000000000..eed5acdbe4
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/utils/PooledClassLoaderEntry.java
@@ -0,0 +1,101 @@
+/*
+ * 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 com.google.common.annotations.VisibleForTesting;
+
+/**
+ * An entry in the {@link ClassLoaderPool} that holds a shared {@link
IsolatedClassLoader} and
+ * tracks its reference count. When the reference count drops to zero, the
ClassLoader can be safely
+ * cleaned up and removed from the pool.
+ *
+ * <p><b>Threading contract:</b> {@link #incrementRefCount()} and {@link
#decrementRefCount()} must
+ * only be called inside {@link
java.util.concurrent.ConcurrentHashMap#compute}, which serializes
+ * access per key. A plain {@code int} is used instead of {@code
AtomicInteger} to make this
+ * contract explicit — misuse outside {@code compute()} will produce data
races that are immediately
+ * visible rather than silently incorrect.
+ */
+public class PooledClassLoaderEntry {
+
+ private final ClassLoaderKey key;
+ private final IsolatedClassLoader classLoader;
+ private int refCount = 1;
+ private boolean cleanedUp = false;
+
+ /**
+ * Constructs a PooledClassLoaderEntry.
+ *
+ * @param key The key identifying this ClassLoader in the pool.
+ * @param classLoader The isolated ClassLoader instance.
+ */
+ public PooledClassLoaderEntry(ClassLoaderKey key, IsolatedClassLoader
classLoader) {
+ this.key = key;
+ this.classLoader = classLoader;
+ }
+
+ /** Returns the key identifying this ClassLoader in the pool. */
+ public ClassLoaderKey key() {
+ return key;
+ }
+
+ /** Returns the isolated ClassLoader instance. */
+ public IsolatedClassLoader classLoader() {
+ return classLoader;
+ }
+
+ /**
+ * Returns the current reference count. Note: outside of {@code
ConcurrentHashMap.compute()}, the
+ * returned value may be stale.
+ */
+ @VisibleForTesting
+ int refCount() {
+ return refCount;
+ }
+
+ /**
+ * Increments the reference count and returns the new value. Must only be
called inside {@link
+ * java.util.concurrent.ConcurrentHashMap#compute}.
+ */
+ int incrementRefCount() {
+ return ++refCount;
+ }
+
+ /**
+ * Decrements the reference count and returns the new value. Must only be
called inside {@link
+ * java.util.concurrent.ConcurrentHashMap#compute}.
+ */
+ int decrementRefCount() {
+ if (refCount <= 0) {
+ throw new IllegalStateException("Cannot decrement refCount below zero
for key: " + key);
+ }
+ return --refCount;
+ }
+
+ /**
+ * Marks this entry as cleaned up. Returns {@code true} if this is the first
call, {@code false}
+ * if cleanup was already performed. Used as a defense-in-depth guard
against double cleanup.
+ */
+ boolean markCleanedUp() {
+ if (cleanedUp) {
+ return false;
+ }
+ cleanedUp = true;
+ return true;
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java
b/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java
new file mode 100644
index 0000000000..6e59027004
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java
@@ -0,0 +1,349 @@
+/*
+ * 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.catalog;
+
+import com.google.common.collect.ImmutableMap;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Map;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.memory.TestMemoryEntityStore;
+import
org.apache.gravitino.storage.memory.TestMemoryEntityStore.InMemoryEntityStore;
+import org.apache.gravitino.utils.ClassLoaderPool;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration tests for ClassLoaderPool with CatalogManager. Tests that
same-type catalogs share a
+ * ClassLoader and that closing one catalog does not affect others of the same
type.
+ */
+public class TestClassLoaderPoolIntegration {
+
+ private static CatalogManager catalogManager;
+ private static InMemoryEntityStore entityStore;
+ private static Config config;
+ private static final String METALAKE = "metalake";
+ private static final String PROVIDER = "test";
+
+ private static BaseMetalake metalakeEntity =
+ BaseMetalake.builder()
+ .withId(1L)
+ .withName(METALAKE)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withVersion(SchemaVersion.V_0_1)
+ .build();
+
+ @BeforeAll
+ public static void setUp() throws IOException, IllegalAccessException {
+ config = new Config(false) {};
+ config.set(Configs.CATALOG_LOAD_ISOLATED, false);
+
+ entityStore = new TestMemoryEntityStore.InMemoryEntityStore();
+ entityStore.initialize(config);
+ entityStore.put(metalakeEntity, true);
+
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new
LockManager(config), true);
+ }
+
+ @AfterAll
+ public static void tearDown() throws IOException {
+ if (catalogManager != null) {
+ catalogManager.close();
+ }
+ if (entityStore != null) {
+ entityStore.close();
+ }
+ }
+
+ @BeforeEach
+ public void beforeEach() throws IOException {
+ catalogManager = new CatalogManager(config, entityStore, new
RandomIdGenerator());
+ }
+
+ @AfterEach
+ public void afterEach() throws IOException {
+ if (catalogManager != null) {
+ catalogManager.close();
+ }
+ entityStore.clear();
+ entityStore.put(metalakeEntity, true);
+ }
+
+ @Test
+ public void testSameTypeCatalogsShareClassLoader() throws
IllegalAccessException {
+ Map<String, String> props =
+ ImmutableMap.of("key1", "value1", "key2", "value2", "key5-1",
"value3");
+
+ Catalog catalog1 =
+ catalogManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog1"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 1",
+ props);
+ Catalog catalog2 =
+ catalogManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog2"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 2",
+ props);
+
+ Assertions.assertNotNull(catalog1);
+ Assertions.assertNotNull(catalog2);
+
+ // Both catalogs should use the same provider which means they share
ClassLoader
+ CatalogManager.CatalogWrapper wrapper1 =
+
catalogManager.getCatalogCache().getIfPresent(NameIdentifier.of(METALAKE,
"catalog1"));
+ CatalogManager.CatalogWrapper wrapper2 =
+
catalogManager.getCatalogCache().getIfPresent(NameIdentifier.of(METALAKE,
"catalog2"));
+
+ Assertions.assertNotNull(wrapper1);
+ Assertions.assertNotNull(wrapper2);
+
+ // Verify they actually share the same ClassLoader instance
+ Object classLoader1 = FieldUtils.readField(wrapper1, "classLoader", true);
+ Object classLoader2 = FieldUtils.readField(wrapper2, "classLoader", true);
+ Assertions.assertSame(
+ classLoader1, classLoader2, "Same-type catalogs should share a
ClassLoader");
+ }
+
+ @Test
+ public void testClosingOneCatalogDoesNotAffectOthers() {
+ Map<String, String> props =
+ ImmutableMap.of("key1", "value1", "key2", "value2", "key5-1",
"value3");
+
+ catalogManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog1"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 1",
+ props);
+ catalogManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog2"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 2",
+ props);
+
+ // Drop catalog1 (shares the pooled ClassLoader with catalog2). Releasing
catalog1 must not
+ // trigger the final-release cleanup while catalog2 still holds a
reference.
+ catalogManager.disableCatalog(NameIdentifier.of(METALAKE, "catalog1"));
+ catalogManager.dropCatalog(NameIdentifier.of(METALAKE, "catalog1"));
+
+ // catalog2 must still be functionally usable after catalog1's drop. Note:
with
+ // CATALOG_LOAD_ISOLATED=false the ClassLoader is an empty in-process
loader, so this is an
+ // end-to-end smoke check that the shared wrapper is not broken by the
drop. The real proof that
+ // the shared ClassLoader/driver is not prematurely cleaned up (which only
matters with real
+ // provider jars) lives in the docker-tagged IcebergClassLoaderPoolIT.
+ NameIdentifier catalog2 = NameIdentifier.of(METALAKE, "catalog2");
+ Schema schema =
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalogManager
+ .loadCatalogAndWrap(catalog2)
+ .doWithSchemaOps(
+ schemaOps ->
+ schemaOps.createSchema(
+ NameIdentifier.of(METALAKE, "catalog2",
"schema1"),
+ "comment",
+ ImmutableMap.of())));
+ Assertions.assertNotNull(schema);
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalogManager
+ .loadCatalogAndWrap(catalog2)
+ .doWithSchemaOps(
+ schemaOps -> schemaOps.listSchemas(Namespace.of(METALAKE,
"catalog2"))));
+ }
+
+ @Test
+ public void testTestConnectionDoesNotBreakLiveCatalogSharingSameKey()
+ throws IllegalAccessException {
+ Map<String, String> props =
+ ImmutableMap.of("key1", "value1", "key2", "value2", "key5-1",
"value3");
+
+ // A live catalog holds a reference to the pooled ClassLoader for this key.
+ NameIdentifier liveIdent = NameIdentifier.of(METALAKE, "live_catalog");
+ catalogManager.createCatalog(
+ liveIdent, Catalog.Type.RELATIONAL, PROVIDER, "live catalog", props);
+
+ CatalogManager.CatalogWrapper liveWrapper =
+ catalogManager.getCatalogCache().getIfPresent(liveIdent);
+ Assertions.assertNotNull(liveWrapper);
+
+ ClassLoaderPool pool =
+ (ClassLoaderPool) FieldUtils.readField(catalogManager,
"classLoaderPool", true);
+ Assertions.assertEquals(1, pool.size(), "one pooled entry for the shared
key");
+
+ // testConnection creates a throwaway wrapper that acquires and then
releases the same key
+ // synchronously (acquire on create, release in its finally). The release
must only decrement
+ // the reference count, not run final cleanup, because the live catalog
still shares the key.
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalogManager.testConnection(
+ NameIdentifier.of(METALAKE, "probe_catalog"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "probe",
+ props));
+
+ // Strong, non-racy check: testConnection's acquire+release is
synchronous, so the pooled entry
+ // for the live catalog's key must survive (size stays 1). A refCount that
wrongly hit 0 would
+ // have removed the entry (size 0) and destroyed the live catalog's
ClassLoader.
+ Assertions.assertEquals(
+ 1, pool.size(), "shared-key entry must survive testConnection's
acquire/release");
+
+ // The live catalog must remain usable after testConnection's
acquire/release cycle.
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalogManager
+ .loadCatalogAndWrap(liveIdent)
+ .doWithSchemaOps(
+ schemaOps -> schemaOps.listSchemas(Namespace.of(METALAKE,
"live_catalog"))));
+ }
+
+ @Test
+ public void testSharingDisabledEachCatalogGetsOwnClassLoader() throws
IllegalAccessException {
+ Config noSharingConfig = new Config(false) {};
+ noSharingConfig.set(Configs.CATALOG_LOAD_ISOLATED, false);
+ noSharingConfig.set(Configs.CATALOG_CLASSLOADER_SHARING_ENABLED, false);
+
+ CatalogManager noSharingManager =
+ new CatalogManager(noSharingConfig, entityStore, new
RandomIdGenerator());
+ try {
+ Map<String, String> props =
+ ImmutableMap.of("key1", "value1", "key2", "value2", "key5-1",
"value3");
+
+ noSharingManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog1"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 1",
+ props);
+ noSharingManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog2"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 2",
+ props);
+
+ CatalogManager.CatalogWrapper wrapper1 =
+
noSharingManager.getCatalogCache().getIfPresent(NameIdentifier.of(METALAKE,
"catalog1"));
+ CatalogManager.CatalogWrapper wrapper2 =
+
noSharingManager.getCatalogCache().getIfPresent(NameIdentifier.of(METALAKE,
"catalog2"));
+
+ Assertions.assertNotNull(wrapper1);
+ Assertions.assertNotNull(wrapper2);
+
+ // With sharing disabled, each catalog should have its own ClassLoader
instance
+ Object classLoader1 = FieldUtils.readField(wrapper1, "classLoader",
true);
+ Object classLoader2 = FieldUtils.readField(wrapper2, "classLoader",
true);
+ Assertions.assertNotSame(
+ classLoader1,
+ classLoader2,
+ "With sharing disabled, catalogs should NOT share a ClassLoader");
+
+ // Non-pooled wrappers should not hold a pool reference
+ Object pool1 = FieldUtils.readField(wrapper1, "pool", true);
+ Assertions.assertNull(pool1, "Non-pooled wrapper should not reference
the pool");
+ } finally {
+ noSharingManager.close();
+ }
+ }
+
+ @Test
+ public void testSharingDisabledDropOneCatalogDoesNotAffectOther() {
+ Config noSharingConfig = new Config(false) {};
+ noSharingConfig.set(Configs.CATALOG_LOAD_ISOLATED, false);
+ noSharingConfig.set(Configs.CATALOG_CLASSLOADER_SHARING_ENABLED, false);
+
+ CatalogManager noSharingManager =
+ new CatalogManager(noSharingConfig, entityStore, new
RandomIdGenerator());
+ try {
+ Map<String, String> props =
+ ImmutableMap.of("key1", "value1", "key2", "value2", "key5-1",
"value3");
+
+ noSharingManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog1"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 1",
+ props);
+ noSharingManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog2"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 2",
+ props);
+
+ // Drop catalog1
+ noSharingManager.disableCatalog(NameIdentifier.of(METALAKE, "catalog1"));
+ noSharingManager.dropCatalog(NameIdentifier.of(METALAKE, "catalog1"));
+
+ // catalog2 should still be loadable
+ Catalog loaded =
+ Assertions.assertDoesNotThrow(
+ () -> noSharingManager.loadCatalog(NameIdentifier.of(METALAKE,
"catalog2")));
+ Assertions.assertNotNull(loaded);
+ } finally {
+ noSharingManager.close();
+ }
+ }
+
+ @Test
+ public void testClassLoaderPoolCleanupOnManagerClose() throws
IllegalAccessException {
+ Map<String, String> props =
+ ImmutableMap.of("key1", "value1", "key2", "value2", "key5-1",
"value3");
+
+ catalogManager.createCatalog(
+ NameIdentifier.of(METALAKE, "catalog1"),
+ Catalog.Type.RELATIONAL,
+ PROVIDER,
+ "test catalog 1",
+ props);
+
+ ClassLoaderPool pool =
+ (ClassLoaderPool) FieldUtils.readField(catalogManager,
"classLoaderPool", true);
+ Assertions.assertNotNull(pool);
+
+ catalogManager.close();
+
+ // After close, the pool should be empty
+ Assertions.assertEquals(0, pool.size());
+ catalogManager = null;
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/utils/TestClassLoaderPool.java
b/core/src/test/java/org/apache/gravitino/utils/TestClassLoaderPool.java
new file mode 100644
index 0000000000..7af4836002
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/utils/TestClassLoaderPool.java
@@ -0,0 +1,428 @@
+/*
+ * 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 com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestClassLoaderPool {
+
+ private IsolatedClassLoader createDummyClassLoader() {
+ return new IsolatedClassLoader(
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList());
+ }
+
+ private ClassLoaderKey simpleKey(String provider) {
+ return new ClassLoaderKey(provider, null);
+ }
+
+ @Test
+ public void testAcquireFirstTimeCreatesNewEntry() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key = simpleKey("iceberg");
+ PooledClassLoaderEntry entry = pool.acquire(key,
this::createDummyClassLoader);
+
+ Assertions.assertNotNull(entry);
+ Assertions.assertEquals(1, entry.refCount());
+ Assertions.assertNotNull(entry.classLoader());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testAcquireSameKeyReusesEntry() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key = simpleKey("iceberg");
+ PooledClassLoaderEntry entry1 = pool.acquire(key,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key,
this::createDummyClassLoader);
+
+ Assertions.assertSame(entry1, entry2);
+ Assertions.assertEquals(2, entry1.refCount());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testDifferentKeysCreateDifferentEntries() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 = simpleKey("iceberg");
+ ClassLoaderKey key2 = simpleKey("hive");
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertEquals(1, entry1.refCount());
+ Assertions.assertEquals(1, entry2.refCount());
+ Assertions.assertEquals(2, pool.size());
+ }
+ }
+
+ @Test
+ public void testReleaseDecrementsRefCount() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key = simpleKey("iceberg");
+ PooledClassLoaderEntry entry = pool.acquire(key,
this::createDummyClassLoader);
+ pool.acquire(key, this::createDummyClassLoader);
+ Assertions.assertEquals(2, entry.refCount());
+
+ pool.release(entry);
+ Assertions.assertEquals(1, entry.refCount());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testReleaseToZeroRemovesEntry() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key = simpleKey("iceberg");
+ PooledClassLoaderEntry entry = pool.acquire(key,
this::createDummyClassLoader);
+ Assertions.assertEquals(1, pool.size());
+
+ pool.release(entry);
+ Assertions.assertEquals(0, pool.size());
+ }
+ }
+
+ @Test
+ public void testDoubleReleaseIsHandledGracefully() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key = simpleKey("iceberg");
+ PooledClassLoaderEntry entry = pool.acquire(key,
this::createDummyClassLoader);
+
+ pool.release(entry); // refCount 1 -> 0, entry removed
+ Assertions.assertEquals(0, pool.size());
+
+ // Second release: entry is already removed from the pool, should log a
warning
+ // but not throw or corrupt state
+ Assertions.assertDoesNotThrow(() -> pool.release(entry));
+ Assertions.assertEquals(0, pool.size());
+
+ // Verify the key can be re-acquired cleanly with no corrupted state
+ PooledClassLoaderEntry fresh = pool.acquire(key,
this::createDummyClassLoader);
+ Assertions.assertNotNull(fresh);
+ Assertions.assertEquals(1, fresh.refCount());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testReleaseOneDoesNotAffectOtherSameType() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key = simpleKey("iceberg");
+ PooledClassLoaderEntry entryA = pool.acquire(key,
this::createDummyClassLoader);
+ pool.acquire(key, this::createDummyClassLoader); // entryB shares same
entry
+ Assertions.assertEquals(2, entryA.refCount());
+
+ pool.release(entryA);
+ Assertions.assertEquals(1, entryA.refCount());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testConcurrentAcquireAndRelease() throws Exception {
+ ClassLoaderPool pool = new ClassLoaderPool();
+ ClassLoaderKey key = simpleKey("iceberg");
+ int threadCount = 20;
+ CyclicBarrier barrier = new CyclicBarrier(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+
+ List<Future<PooledClassLoaderEntry>> futures = new ArrayList<>();
+ for (int i = 0; i < threadCount; i++) {
+ futures.add(
+ executor.submit(
+ () -> {
+ barrier.await();
+ return pool.acquire(key, this::createDummyClassLoader);
+ }));
+ }
+
+ List<PooledClassLoaderEntry> entries = new ArrayList<>();
+ for (Future<PooledClassLoaderEntry> future : futures) {
+ entries.add(future.get(5, TimeUnit.SECONDS));
+ }
+
+ // All should reference the same entry
+ PooledClassLoaderEntry first = entries.get(0);
+ for (PooledClassLoaderEntry entry : entries) {
+ Assertions.assertSame(first, entry);
+ }
+ Assertions.assertEquals(threadCount, first.refCount());
+ Assertions.assertEquals(1, pool.size());
+
+ // Now release all concurrently
+ CyclicBarrier releaseBarrier = new CyclicBarrier(threadCount);
+ List<Future<?>> releaseFutures = new ArrayList<>();
+ for (int i = 0; i < threadCount; i++) {
+ releaseFutures.add(
+ executor.submit(
+ () -> {
+ releaseBarrier.await();
+ pool.release(first);
+ return null;
+ }));
+ }
+
+ for (Future<?> future : releaseFutures) {
+ future.get(5, TimeUnit.SECONDS);
+ }
+
+ Assertions.assertEquals(0, pool.size());
+
+ executor.shutdown();
+ pool.close();
+ }
+
+ @Test
+ public void testCloseCleanupAllEntries() {
+ ClassLoaderPool pool = new ClassLoaderPool();
+ ClassLoaderKey key1 = simpleKey("iceberg");
+ ClassLoaderKey key2 = simpleKey("hive");
+ pool.acquire(key1, this::createDummyClassLoader);
+ pool.acquire(key2, this::createDummyClassLoader);
+ Assertions.assertEquals(2, pool.size());
+
+ pool.close();
+ Assertions.assertEquals(0, pool.size());
+ }
+
+ @Test
+ public void testKeyWithPackageProperty() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 = new ClassLoaderKey("iceberg",
ImmutableMap.of("package", "/path/a"));
+ ClassLoaderKey key2 = new ClassLoaderKey("iceberg",
ImmutableMap.of("package", "/path/b"));
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertEquals(2, pool.size());
+ }
+ }
+
+ @Test
+ public void testKeyWithAuthorizationProvider() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 =
+ new ClassLoaderKey("iceberg",
ImmutableMap.of("authorization-provider", "ranger"));
+ ClassLoaderKey key2 = simpleKey("iceberg");
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertEquals(2, pool.size());
+ }
+ }
+
+ @Test
+ public void testDifferentKerberosPrincipalsCreateDifferentEntries() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ Map<String, String> props1 =
+ ImmutableMap.of(
+ "authentication.type", "kerberos",
+ "authentication.kerberos.principal", "user-A@REALM",
+ "authentication.kerberos.keytab-uri", "/keytabs/a.keytab");
+ Map<String, String> props2 =
+ ImmutableMap.of(
+ "authentication.type", "kerberos",
+ "authentication.kerberos.principal", "user-B@REALM",
+ "authentication.kerberos.keytab-uri", "/keytabs/b.keytab");
+
+ ClassLoaderKey key1 = new ClassLoaderKey("iceberg", props1);
+ ClassLoaderKey key2 = new ClassLoaderKey("iceberg", props2);
+ ClassLoaderKey key3 = simpleKey("iceberg");
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry3 = pool.acquire(key3,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertNotSame(entry1, entry3);
+ Assertions.assertNotSame(entry2, entry3);
+ Assertions.assertEquals(3, pool.size());
+ }
+ }
+
+ @Test
+ public void testSameKerberosPrincipalSharesEntry() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ Map<String, String> props =
+ ImmutableMap.of(
+ "authentication.type", "kerberos",
+ "authentication.kerberos.principal", "user-A@REALM",
+ "authentication.kerberos.keytab-uri", "/keytabs/a.keytab");
+
+ ClassLoaderKey key1 = new ClassLoaderKey("iceberg", props);
+ ClassLoaderKey key2 = new ClassLoaderKey("iceberg", props);
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertSame(entry1, entry2);
+ Assertions.assertEquals(2, entry1.refCount());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testDifferentMetastoreUrisCreateDifferentEntries() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 =
+ new ClassLoaderKey(
+ "lakehouse-iceberg", ImmutableMap.of("metastore.uris",
"thrift://hms-A:9083"));
+ ClassLoaderKey key2 =
+ new ClassLoaderKey(
+ "lakehouse-iceberg", ImmutableMap.of("metastore.uris",
"thrift://hms-B:9083"));
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertEquals(2, pool.size());
+ }
+ }
+
+ @Test
+ public void testSameMetastoreUrisShareEntry() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 =
+ new ClassLoaderKey(
+ "lakehouse-iceberg", ImmutableMap.of("metastore.uris",
"thrift://hms-A:9083"));
+ ClassLoaderKey key2 =
+ new ClassLoaderKey(
+ "lakehouse-iceberg", ImmutableMap.of("metastore.uris",
"thrift://hms-A:9083"));
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertSame(entry1, entry2);
+ Assertions.assertEquals(2, entry1.refCount());
+ Assertions.assertEquals(1, pool.size());
+ }
+ }
+
+ @Test
+ public void testDifferentJdbcUrlsCreateDifferentEntries() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 =
+ new ClassLoaderKey(
+ "jdbc-mysql", ImmutableMap.of("jdbc-url",
"jdbc:mysql://host-A:3306/db"));
+ ClassLoaderKey key2 =
+ new ClassLoaderKey(
+ "jdbc-mysql", ImmutableMap.of("jdbc-url",
"jdbc:mysql://host-B:3306/db"));
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertEquals(2, pool.size());
+ }
+ }
+
+ @Test
+ public void testDifferentDefaultFsCreateDifferentEntries() {
+ try (ClassLoaderPool pool = new ClassLoaderPool()) {
+ ClassLoaderKey key1 =
+ new ClassLoaderKey(
+ "lakehouse-iceberg",
+ ImmutableMap.of(
+ "metastore.uris", "thrift://hms:9083",
+ "fs.defaultFS", "hdfs://cluster-A:8020"));
+ ClassLoaderKey key2 =
+ new ClassLoaderKey(
+ "lakehouse-iceberg",
+ ImmutableMap.of(
+ "metastore.uris", "thrift://hms:9083",
+ "fs.defaultFS", "hdfs://cluster-B:8020"));
+
+ PooledClassLoaderEntry entry1 = pool.acquire(key1,
this::createDummyClassLoader);
+ PooledClassLoaderEntry entry2 = pool.acquire(key2,
this::createDummyClassLoader);
+
+ Assertions.assertNotSame(entry1, entry2);
+ Assertions.assertEquals(2, pool.size());
+ }
+ }
+
+ @Test
+ public void testAcquireAfterCloseThrows() {
+ ClassLoaderPool pool = new ClassLoaderPool();
+ pool.close();
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> pool.acquire(simpleKey("iceberg"),
this::createDummyClassLoader));
+ }
+
+ @Test
+ public void testConcurrentAcquireDuringClose() throws Exception {
+ int threadCount = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ ClassLoaderPool pool = new ClassLoaderPool();
+
+ // Pre-populate the pool with entries that have active refCounts > 0
+ // to exercise the force-cleanup path in close()
+ ClassLoaderKey heldKey = simpleKey("held");
+ pool.acquire(heldKey, this::createDummyClassLoader); // refCount=1, never
released
+
+ // Race: some threads try to acquire while another thread closes
+ CyclicBarrier barrier = new CyclicBarrier(threadCount + 1);
+ List<Future<?>> futures = new ArrayList<>();
+ for (int i = 0; i < threadCount; i++) {
+ final int idx = i;
+ futures.add(
+ executor.submit(
+ () -> {
+ barrier.await();
+ ClassLoaderKey k = simpleKey("type-" + idx);
+ try {
+ PooledClassLoaderEntry e = pool.acquire(k,
this::createDummyClassLoader);
+ // If acquire succeeded, release it
+ pool.release(e);
+ } catch (IllegalStateException expected) {
+ // Pool was closed before or during acquire — expected
+ }
+ return null;
+ }));
+ }
+
+ // Close from the main thread racing with acquires — this force-cleans
+ // the held entry (refCount > 0) and drains any concurrent insertions
+ barrier.await();
+ pool.close();
+
+ for (Future<?> future : futures) {
+ future.get(5, TimeUnit.SECONDS);
+ }
+
+ // After close + all threads done, pool must be empty
+ Assertions.assertEquals(0, pool.size());
+ executor.shutdown();
+ }
+}
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index 14ba65d06a..242ac37dee 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -159,6 +159,7 @@ The Gravitino server uses a tree lock to ensure data
consistency. The tree lock
|----------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------|---------------|
| `gravitino.catalog.cache.evictionIntervalMs` | The interval in
milliseconds to evict the catalog cache; default 3600000ms(1h).
| `3600000` | No |
0.1.0 |
| `gravitino.catalog.classloader.isolated` | Whether to use an
isolated classloader for catalog. If `true`, an isolated classloader loads all
catalog-related libraries and configurations, not the AppClassLoader. The
default value is `true`.
| `true` | No
| 0.1.0 |
+| `gravitino.catalog.classloader.sharing.enabled` | Whether to share
one ClassLoader across catalogs that have identical isolation-relevant
properties. When `true` (default), such catalogs reuse a single ClassLoader,
reducing Metaspace usage; when `false`, each catalog gets its own ClassLoader
as in releases before 2.0.0.
| `true`
| No | 2.0.0 |
| `gravitino.catalog.credential.backfillToProperties` | For backward
compatibility only: if `true`, the server exposes hidden catalog credentials
(such as jdbc-user and jdbc-password) in the catalog properties response so
that connectors that do not support credential vending can still read them.
**Enabling this is a security risk** — credentials are visible to anyone who
can read catalog properties. Disable once all connectors are upgraded. |
`false` | No | 1.3.0 |
### Schema configuration
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
index dc74534896..8d2196ecca 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
@@ -19,8 +19,6 @@
package org.apache.gravitino.iceberg.common.ops;
import com.google.common.base.Preconditions;
-import java.sql.Driver;
-import java.sql.DriverManager;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
@@ -34,7 +32,6 @@ import
org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
import org.apache.gravitino.iceberg.common.cache.TableMetadataCache;
import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.gravitino.utils.ClassUtils;
-import org.apache.gravitino.utils.IsolatedClassLoader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.iceberg.BaseTable;
@@ -81,7 +78,6 @@ public class IcebergCatalogWrapper implements AutoCloseable {
private volatile SupportsNamespaces asNamespaceCatalog;
private final IcebergCatalogBackend catalogBackend;
private final IcebergConfig icebergConfig;
- private String catalogUri = null;
private volatile TableMetadataCache metadataCache;
private final Configuration configuration;
@@ -97,9 +93,6 @@ public class IcebergCatalogWrapper implements AutoCloseable {
throw new IllegalArgumentException("The 'warehouse' parameter must
have a value.");
}
}
- if (!IcebergCatalogBackend.MEMORY.equals(catalogBackend)) {
- this.catalogUri = icebergConfig.get(IcebergConfig.CATALOG_URI);
- }
Map<String, String> catalogPropertiesMap =
icebergConfig.getIcebergCatalogProperties();
this.configuration = FileSystemUtils.createConfiguration(null,
catalogPropertiesMap);
}
@@ -405,79 +398,12 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
if (cache != null) {
cache.close();
}
-
- // For Iceberg REST server which use the same classloader when recreating
catalog wrapper, the
- // Driver couldn't be reloaded after deregister()
- if (useDifferentClassLoader()) {
- closeJdbcDriverResources();
- }
}
public boolean isRESTCatalog() {
return getCatalog() instanceof RESTCatalog;
}
- /**
- * Whether the wrapper is recreated with a different classloader.
- *
- * <p>Returning {@code true} allows JDBC drivers loaded by an isolated
classloader to be
- * deregistered when the wrapper closes so the classloader can be garbage
collected. Implementors
- * that intentionally reuse the same classloader (for example, an Iceberg
REST server instance)
- * should override and return {@code false} to skip deregistration.
- */
- protected boolean useDifferentClassLoader() {
- return true;
- }
-
- private void closeJdbcDriverResources() {
- // Because each catalog in Gravitino has its own classloader, after a
catalog is no longer used
- // for a long time or dropped, the instance of classloader needs to be
released. In order to
- // let JVM GC remove the classloader, we need to release the resources of
the classloader. The
- // resources include the driver of the catalog backend and the
- // AbandonedConnectionCleanupThread of MySQL. For more information about
- // AbandonedConnectionCleanupThread, please refer to the corresponding
java doc of MySQL
- // driver.
- if (catalogUri != null && catalogUri.contains("mysql")) {
- closeMySQLCatalogResource();
- } else if (catalogUri != null && catalogUri.contains("postgresql")) {
- closePostgreSQLCatalogResource();
- }
- }
-
- private void closeMySQLCatalogResource() {
- try {
- // Close thread AbandonedConnectionCleanupThread if we are using
`com.mysql.cj.jdbc.Driver`,
- // for driver `com.mysql.jdbc.Driver` (deprecated), the daemon thread
maybe not this one.
- Class.forName("com.mysql.cj.jdbc.AbandonedConnectionCleanupThread")
- .getMethod("uncheckedShutdown")
- .invoke(null);
- LOG.info("AbandonedConnectionCleanupThread has been shutdown...");
-
- // Unload the MySQL driver, only Unload the driver if it is loaded by
- // IsolatedClassLoader.
- closeDriverLoadedByIsolatedClassLoader(catalogUri);
- } catch (Exception e) {
- LOG.warn("Failed to shutdown AbandonedConnectionCleanupThread or
deregister MySQL driver", e);
- }
- }
-
- private void closeDriverLoadedByIsolatedClassLoader(String uri) {
- try {
- Driver driver = DriverManager.getDriver(uri);
- if (driver.getClass().getClassLoader().getClass()
- == IsolatedClassLoader.CUSTOM_CLASS_LOADER_CLASS) {
- DriverManager.deregisterDriver(driver);
- LOG.info("Driver {} has been deregistered...", driver);
- }
- } catch (Exception e) {
- LOG.warn("Failed to deregister driver", e);
- }
- }
-
- private void closePostgreSQLCatalogResource() {
- closeDriverLoadedByIsolatedClassLoader(catalogUri);
- }
-
@Getter
@Setter
public static final class IcebergTableChange {
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index a2749325ae..1af51f5295 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -271,11 +271,6 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
}
- @Override
- protected boolean useDifferentClassLoader() {
- return false;
- }
-
@VisibleForTesting
protected LoadTableResponse injectCredentialConfig(
TableIdentifier tableIdentifier,
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
index 8e7db655b6..e127c0bf6f 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
@@ -63,6 +63,20 @@ public enum TestDatabaseName {
/** Represents the MySQL database used for testing catalog credential
integration. */
MYSQL_CATALOG_CREDENTIAL_IT,
+ /**
+ * Represents the first MySQL-backed database used by
+ * lakehouse.iceberg.integration.test.IcebergClassLoaderPoolIT to verify
ClassLoader isolation and
+ * sharing across Iceberg catalogs whose JDBC backend URI differs only by
database name.
+ */
+ MYSQL_TEST_ICEBERG_CLASSLOADER_POOL_A,
+
+ /**
+ * Represents the second MySQL-backed database used by
+ * lakehouse.iceberg.integration.test.IcebergClassLoaderPoolIT (different
{@code uri} than {@link
+ * #MYSQL_TEST_ICEBERG_CLASSLOADER_POOL_A}).
+ */
+ MYSQL_TEST_ICEBERG_CLASSLOADER_POOL_B,
+
PG_JDBC_BACKEND,
/** Represents the PostgreSQL database for CatalogPostgreSqlIT. */
