yuqi1129 commented on code in PR #10480:
URL: https://github.com/apache/gravitino/pull/10480#discussion_r3535110304
##########
core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java:
##########
@@ -1153,9 +1237,29 @@ CatalogWrapper createCatalogWrapper(
* @return The resolved properties.
*/
private Map<String, String> getResolvedProperties(CatalogEntity entity) {
- CatalogWrapper catalogWrapper =
loadCatalogAndWrap(entity.nameIdentifier());
- return catalogWrapper.classLoader.withClassLoader(
- cl -> catalogWrapper.catalog.properties(), RuntimeException.class);
+ Map<String, String> conf = entity.getProperties();
+ String provider = entity.getProvider();
+
+ if (!classLoaderSharingEnabled) {
+ IsolatedClassLoader classLoader = createClassLoader(provider, conf);
+ try {
+ BaseCatalog<?> catalog = createBaseCatalog(classLoader, entity);
+ return classLoader.withClassLoader(cl -> catalog.properties(),
RuntimeException.class);
+ } finally {
+ ClassLoaderPool.cleanupClassLoader(classLoader);
+ }
+ }
+
+ ClassLoaderKey key = buildClassLoaderKey(provider, conf);
+ PooledClassLoaderEntry poolEntry =
+ classLoaderPool.acquire(key, () -> createClassLoader(provider, conf));
+ try {
+ IsolatedClassLoader classLoader = poolEntry.classLoader();
+ BaseCatalog<?> catalog = createBaseCatalog(classLoader, entity);
Review Comment:
This `BaseCatalog` (and the non-pooled branch at L1246) is created but never
`close()`d. `BaseCatalog.close()` is what closes `authorizationPlugin`, so for
any catalog with `authorization-provider` set, each `listCatalogsInfo()` call
(L525 maps this over every catalog) leaks a plugin instance. The pooled branch
also builds + tears down a fresh `IsolatedClassLoader` per uncached catalog on
every list call. The previous impl reused the cached `loadCatalogAndWrap(...)`
wrapper — suggest reusing it, or wrapping in try/finally with `catalog.close()`.
##########
core/src/main/java/org/apache/gravitino/utils/ClassLoaderPool.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.
+ */
+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();
Review Comment:
Minor: the classloader `factory` here (jar scanning) and `doFinalCleanup()`
(thread interruption + reflection over all threads) both run inside
`ConcurrentHashMap.compute`, holding the bin lock during heavy work. Fine for a
small pool, just flagging for potential create/drop latency under concurrency.
##########
core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java:
##########
@@ -0,0 +1,281 @@
+/*
+ * 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.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);
Review Comment:
Testing gap: these tests run with `CATALOG_LOAD_ISOLATED=false` and empty
dummy classloaders, so the two core claims of this PR — actual Metaspace
savings and no cross-catalog interference from a *shared* classloader — aren't
exercised. Suggest adding docker-tagged ITs with real provider jars:
- **GC/leak**: hold a `WeakReference` to the `IsolatedClassLoader`, drop the
catalog, `System.gc()`, assert it is collected (catches leftover ThreadLocals /
lingering threads / undereg'd drivers).
- **testConnection release**: with a live catalog sharing the key,
`testConnection` should acquire+release without dropping refCount to 0 and
breaking the live catalog.
- **Key blind spot**: Iceberg's JDBC backend uses `uri`, not `jdbc-url` (not
in `DEFAULT_ISOLATION_PROPERTY_KEYS`) — verify two mysql-backed Iceberg
catalogs sharing a classloader don't cross-contaminate.
- **iceberg-rest-server regression**: the removed
`useDifferentClassLoader()` override — verify REST-server JDBC-backend wrapper
rebuild still works.
##########
core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java:
##########
@@ -256,7 +291,14 @@ public void close() {
LOG.warn("Failed to close catalog", e);
}
- classLoader.close();
+ 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);
Review Comment:
Unlike the pooled branch above (which nulls `poolEntry`, making a repeat
close a no-op), this non-pooled branch has no idempotency guard: a second
`close()` re-runs `cleanupClassLoader()` (double driver-deregister / resource
cleanup / `classLoader.close()`). Since `close()` was made `synchronized`,
consider a `closed` flag so both paths are symmetric.
##########
core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java:
##########
@@ -0,0 +1,281 @@
+/*
+ * 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.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() {
Review Comment:
For real coverage this should issue an actual request on `catalog2` after
dropping `catalog1` (with real provider jars), not just `loadCatalog`. The
cleanup-on-final-release path (JDBC driver deregister,
`AbandonedConnectionCleanupThread` shutdown) can only be caught by exercising
`catalog2` after `catalog1`'s drop when they share a classloader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
