This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 9c4fd069af [Cherry-pick to branch-1.3] [#12851] fix(iceberg-rest):
Share the managed memory catalog in auxiliary mode (#12852) (#12923)
9c4fd069af is described below
commit 9c4fd069afe9214650687ebb0f88cd47b177e880
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 4 18:47:30 2026 +0800
[Cherry-pick to branch-1.3] [#12851] fix(iceberg-rest): Share the managed
memory catalog in auxiliary mode (#12852) (#12923)
**Cherry-pick Information:**
- Original commit: 30687f68ff8c2a276a4a0fbf13211435da36f7f9
- Target branch: `branch-1.3`
- Status: ✅ **Conflicts resolved**
**Resolution:**
- Preserved the existing `branch-1.3` authentication and credential
handling.
- Applied the catalog UUID, shared in-memory catalog, and catalog-drop
cleanup changes.
- Removed all committed conflict markers without bringing in unrelated
main-only changes.
**Validation:**
- `./gradlew spotlessApply`
- `./gradlew :iceberg:iceberg-common:test --tests
org.apache.gravitino.iceberg.common.utils.TestIcebergCatalogUtil --tests
org.apache.gravitino.iceberg.common.ops.TestIcebergCatalogWrapper`
- `./gradlew :catalogs:catalog-lakehouse-iceberg:test --tests
org.apache.gravitino.catalog.lakehouse.iceberg.TestIcebergCatalogOperations
--tests
org.apache.gravitino.catalog.lakehouse.iceberg.TestIcebergSchema`
- `./gradlew :iceberg:iceberg-rest-server:test --tests
org.apache.gravitino.iceberg.service.provider.TestDynamicIcebergConfigProvider
--tests
org.apache.gravitino.iceberg.service.TestIcebergCatalogWrapperManagerForREST`
- `./gradlew :core:test --tests
org.apache.gravitino.catalog.TestCatalogManager`
- Conflict-marker scan and `git diff --check`
---------
Co-authored-by: roryqi <[email protected]>
Co-authored-by: roryqi <[email protected]>
---
.../lakehouse/iceberg/IcebergConstants.java | 3 +
.../iceberg/IcebergCatalogOperations.java | 16 +++-
.../iceberg/TestIcebergCatalogOperations.java | 24 ++++++
.../lakehouse/iceberg/TestIcebergSchema.java | 3 +-
.../apache/gravitino/catalog/CatalogManager.java | 12 +++
.../gravitino/connector/CatalogDropAware.java | 29 ++++++++
.../gravitino/catalog/TestCatalogManager.java | 13 ++++
.../iceberg/common/ops/IcebergCatalogWrapper.java | 6 +-
.../iceberg/common/utils/IcebergCatalogUtil.java | 35 +++++++++
.../common/ops/TestIcebergCatalogWrapper.java | 27 +++++++
.../common/utils/TestIcebergCatalogUtil.java | 86 ++++++++++++++++++++++
.../service/IcebergCatalogWrapperManager.java | 4 +
.../provider/DynamicIcebergConfigProvider.java | 4 +-
.../TestIcebergCatalogWrapperManagerForREST.java | 46 +++++++++++-
.../provider/TestDynamicIcebergConfigProvider.java | 32 ++++++++
15 files changed, 334 insertions(+), 6 deletions(-)
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
index 354153b303..2d29e9e754 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
@@ -24,6 +24,9 @@ public class IcebergConstants {
public static final String CATALOG_BACKEND = "catalog-backend";
public static final String CATALOG_BACKEND_IMPL = "catalog-backend-impl";
+ /** Internal property containing the unique identifier of a Gravitino
catalog. */
+ public static final String CATALOG_UUID = "catalog_uuid";
+
public static final String GRAVITINO_JDBC_USER = "jdbc-user";
public static final String ICEBERG_JDBC_USER = "jdbc.user";
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 457e5cbd55..d881404ef1 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
@@ -31,12 +31,14 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.SchemaChange;
import
org.apache.gravitino.catalog.lakehouse.iceberg.ops.IcebergCatalogWrapperHelper;
+import org.apache.gravitino.connector.CatalogDropAware;
import org.apache.gravitino.connector.CatalogInfo;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.HasPropertyMetadata;
@@ -56,6 +58,7 @@ import
org.apache.gravitino.iceberg.common.authentication.SupportsKerberos;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper.IcebergTableChange;
import
org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy;
+import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.rel.Column;
@@ -90,7 +93,7 @@ import org.slf4j.LoggerFactory;
/** Operations for interacting with an Apache Iceberg catalog in Apache
Gravitino. */
public class IcebergCatalogOperations
- implements CatalogOperations, SupportsSchemas, TableCatalog, ViewCatalog {
+ implements CatalogOperations, SupportsSchemas, TableCatalog, ViewCatalog,
CatalogDropAware {
private static final String ICEBERG_TABLE_DOES_NOT_EXIST_MSG = "Iceberg
table does not exist: %s";
@@ -98,6 +101,7 @@ public class IcebergCatalogOperations
@VisibleForTesting IcebergCatalogWrapper icebergCatalogWrapper;
+ @VisibleForTesting @Nullable String catalogUuid;
private IcebergCatalogWrapperHelper icebergCatalogWrapperHelper;
private IcebergViewCatalogOperations icebergViewCatalogOperations;
@@ -122,7 +126,8 @@ public class IcebergCatalogOperations
Map<String, String> resultConf = Maps.newHashMap(prefixMap);
resultConf.putAll(gravitinoConfig);
- resultConf.put("catalog_uuid", info.id().toString());
+ this.catalogUuid = info.id().toString();
+ resultConf.put(IcebergConstants.CATALOG_UUID, catalogUuid);
IcebergConfig icebergConfig = new IcebergConfig(resultConf);
IcebergCatalogWrapper rawWrapper = new
IcebergCatalogWrapper(icebergConfig);
@@ -137,6 +142,13 @@ public class IcebergCatalogOperations
this.icebergViewCatalogOperations = new
IcebergViewCatalogOperations(icebergCatalogWrapper);
}
+ @Override
+ public void onCatalogDropped() {
+ if (catalogUuid != null) {
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ }
+ }
+
/** Closes the Iceberg catalog and releases the associated client pool. */
@Override
public void close() {
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
index dfbd129a23..4844452f08 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
@@ -28,7 +28,9 @@ import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.iceberg.rest.responses.ListNamespacesResponse;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -95,4 +97,26 @@ public class TestIcebergCatalogOperations {
Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"db1".equals(id.name())));
Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"db2".equals(id.name())));
}
+
+ @Test
+ public void testCatalogDropRemovesInternedMemoryCatalog() {
+ String catalogUuid = "catalog-uuid";
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ catalogUuid));
+ Object original =
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ IcebergCatalogOperations catalogOperations = new
IcebergCatalogOperations();
+ catalogOperations.catalogUuid = catalogUuid;
+ catalogOperations.onCatalogDropped();
+
+ Object replacement =
+ IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY,
config);
+ Assertions.assertNotSame(original, replacement);
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ }
}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
index 9c9f6f5de5..c028c24b2e 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
@@ -187,7 +187,8 @@ public class TestIcebergSchema {
private IcebergCatalog initIcebergCatalog(String name) {
CatalogEntity entity =
CatalogEntity.builder()
- .withId(1L)
+ // Each test creates a different catalog, so it must also use a
distinct catalog ID.
+ .withId(Integer.toUnsignedLong(name.hashCode()))
.withName(name)
.withNamespace(Namespace.of(META_LAKE_NAME))
.withType(IcebergCatalog.Type.RELATIONAL)
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 d7e2739f07..64acb6ff55 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -79,6 +79,7 @@ import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.connector.CatalogDropAware;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.HasPropertyMetadata;
import org.apache.gravitino.connector.SupportsSchemas;
@@ -1011,6 +1012,17 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
boolean deleted = store.delete(ident, EntityType.CATALOG, true);
if (deleted) {
markLocalMutation(ident);
+ try {
+ catalogWrapper.doWithCatalogOps(
+ operations -> {
+ if (operations instanceof CatalogDropAware) {
+ ((CatalogDropAware) operations).onCatalogDropped();
+ }
+ return null;
+ });
+ } catch (Exception e) {
+ LOG.warn("Failed to clean up resources for dropped catalog
{}", ident, e);
+ }
}
catalogCache.invalidate(ident);
return deleted;
diff --git
a/core/src/main/java/org/apache/gravitino/connector/CatalogDropAware.java
b/core/src/main/java/org/apache/gravitino/connector/CatalogDropAware.java
new file mode 100644
index 0000000000..2972808c1c
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/connector/CatalogDropAware.java
@@ -0,0 +1,29 @@
+/*
+ * 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.connector;
+
+import org.apache.gravitino.annotation.Evolving;
+
+/** Supports cleanup that must run only when a catalog is permanently dropped.
*/
+@Evolving
+public interface CatalogDropAware {
+
+ /** Performs cleanup after the catalog metadata has been permanently
dropped. */
+ void onCatalogDropped();
+}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 4424dc848f..1be48b0d3a 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -57,6 +57,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.connector.CatalogDropAware;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.connector.capability.CapabilityResult;
@@ -1108,10 +1109,21 @@ public class TestCatalogManager {
Mockito.mock(CatalogManager.CatalogWrapper.class,
Mockito.RETURNS_DEEP_STUBS);
Capability capability = Mockito.mock(Capability.class);
CapabilityResult unsupportedResult = CapabilityResult.unsupported("Not
managed");
+ CatalogOperations operations =
+ Mockito.mock(
+ CatalogOperations.class,
+ Mockito.withSettings().extraInterfaces(CatalogDropAware.class));
Mockito.doReturn(catalogWrapper).when(catalogManager).loadCatalogAndWrap(ident);
Mockito.doReturn(catalog).when(catalogWrapper).catalog();
Mockito.doReturn(capability).when(catalogWrapper).capabilities();
Mockito.doReturn(unsupportedResult).when(capability).managedStorage(any());
+ Mockito.doAnswer(
+ invocation -> {
+ ThrowableFunction<CatalogOperations, ?> function =
invocation.getArgument(0);
+ return function.apply(operations);
+ })
+ .when(catalogWrapper)
+ .doWithCatalogOps(any());
catalogManager.getCatalogCache().put(ident, catalogWrapper);
boolean dropped = catalogManager.dropCatalog(ident);
@@ -1119,6 +1131,7 @@ public class TestCatalogManager {
Assertions.assertTrue(dropped);
Assertions.assertFalse(entityStore.exists(ident, EntityType.CATALOG));
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(ident));
+ Mockito.verify((CatalogDropAware) operations).onCatalogDropped();
}
@Test
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 529c352cae..566c5289d0 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
@@ -29,6 +29,7 @@ import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
import org.apache.gravitino.iceberg.common.cache.TableMetadataCache;
@@ -390,7 +391,10 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
} else {
LOG.info("Closing IcebergCatalogWrapper before catalog is initialized");
}
- if (loadedCatalog instanceof AutoCloseable) {
+ boolean internedMemoryCatalog =
+ catalogBackend == IcebergCatalogBackend.MEMORY
+ &&
icebergConfig.getAllConfig().containsKey(IcebergConstants.CATALOG_UUID);
+ if (!internedMemoryCatalog && loadedCatalog instanceof AutoCloseable) {
// JdbcCatalog and ClosableHiveCatalog implement AutoCloseable and will
handle their own
// cleanup
((AutoCloseable) loadedCatalog).close();
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
index 6c12e60455..f849b77ad5 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
@@ -23,11 +23,13 @@ import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
+import java.io.IOException;
import java.sql.SQLException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.exceptions.ConnectionFailedException;
@@ -56,7 +58,19 @@ public class IcebergCatalogUtil {
private static final Logger LOG =
LoggerFactory.getLogger(IcebergCatalogUtil.class);
+ private static final ConcurrentHashMap<String, InMemoryCatalog>
MEMORY_CATALOGS =
+ new ConcurrentHashMap<>();
+
private static InMemoryCatalog loadMemoryCatalog(IcebergConfig
icebergConfig) {
+ String catalogUuid =
icebergConfig.getAllConfig().get(IcebergConstants.CATALOG_UUID);
+ if (catalogUuid == null) {
+ return createMemoryCatalog(icebergConfig);
+ }
+ return MEMORY_CATALOGS.computeIfAbsent(
+ catalogUuid, ignored -> createMemoryCatalog(icebergConfig));
+ }
+
+ private static InMemoryCatalog createMemoryCatalog(IcebergConfig
icebergConfig) {
String icebergCatalogName = icebergConfig.getCatalogBackendName();
InMemoryCatalog memoryCatalog = new
MemoryCatalogWithMetadataLocationSupport();
Map<String, String> resultProperties =
icebergConfig.getIcebergCatalogProperties();
@@ -68,6 +82,27 @@ public class IcebergCatalogUtil {
return memoryCatalog;
}
+ /**
+ * Removes the in-memory Iceberg catalog associated with a permanently
dropped Gravitino catalog.
+ *
+ * @param catalogUuid the unique Gravitino catalog identifier
+ */
+ public static void removeMemoryCatalog(String catalogUuid) {
+ InMemoryCatalog memoryCatalog = MEMORY_CATALOGS.remove(catalogUuid);
+ if (memoryCatalog != null) {
+ try {
+ memoryCatalog.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close dropped in-memory Iceberg catalog {}",
catalogUuid, e);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static void clearMemoryCatalogs() {
+ MEMORY_CATALOGS.clear();
+ }
+
private static HiveCatalog loadHiveCatalog(IcebergConfig icebergConfig) {
ClosableHiveCatalog hiveCatalog = new
HiveCatalogWithMetadataLocationSupport();
HdfsConfiguration hdfsConfiguration = new HdfsConfiguration();
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
index e34f1becb3..e6928e56a8 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
@@ -30,8 +30,12 @@ import
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
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.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -57,6 +61,29 @@ public class TestIcebergCatalogWrapper {
});
}
+ @Test
+ public void testMemoryCatalogSurvivesWrapperClose() throws Exception {
+ String catalogUuid = "catalog-1";
+ IcebergConfig config =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ catalogUuid));
+ IcebergCatalogWrapper firstWrapper = new IcebergCatalogWrapper(config);
+ Catalog firstCatalog = firstWrapper.getCatalog();
+ Namespace namespace = Namespace.of("preserved");
+
firstWrapper.createNamespace(CreateNamespaceRequest.builder().withNamespace(namespace).build());
+
+ firstWrapper.close();
+ IcebergCatalogWrapper secondWrapper = new IcebergCatalogWrapper(config);
+
+ Assertions.assertSame(firstCatalog, secondWrapper.getCatalog());
+ Assertions.assertTrue(secondWrapper.namespaceExists(namespace));
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ }
+
@Test
public void testMetadataCacheShouldInitializeOnFirstAccessAndClose(@TempDir
Path warehouseDir)
throws Exception {
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
index ade4237456..89cf6f7897 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
@@ -36,6 +36,7 @@ import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.jdbc.JdbcCatalog;
import org.apache.iceberg.jdbc.JdbcCatalogWithMetadataLocationSupport;
import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -44,6 +45,91 @@ public class TestIcebergCatalogUtil {
@TempDir private Path warehouse;
+ @AfterEach
+ void tearDown() {
+ IcebergCatalogUtil.clearMemoryCatalogs();
+ }
+
+ @Test
+ void testMemoryCatalogIsInternedByCatalogUuid() {
+ Map<String, String> properties =
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND, "memory",
IcebergConstants.CATALOG_UUID, "catalog-1");
+ IcebergConfig config = new IcebergConfig(properties);
+
+ InMemoryCatalog first =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+ Namespace namespace = Namespace.of("shared");
+ first.createNamespace(namespace);
+ InMemoryCatalog second =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ Assertions.assertSame(first, second);
+ Assertions.assertTrue(second.namespaceExists(namespace));
+ }
+
+ @Test
+ void testMemoryCatalogsWithDifferentUuidsAreIsolated() {
+ IcebergConfig firstConfig =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ "catalog-1"));
+ IcebergConfig secondConfig =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ "catalog-2"));
+
+ Catalog first =
+ IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY,
firstConfig);
+ Catalog second =
+ IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY,
secondConfig);
+
+ Assertions.assertNotSame(first, second);
+ }
+
+ @Test
+ void testMemoryCatalogWithoutUuidIsNotInterned() {
+ IcebergConfig config = new
IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory"));
+
+ Catalog first =
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+ Catalog second =
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ Assertions.assertNotSame(first, second);
+ }
+
+ @Test
+ void testDroppedMemoryCatalogIsRemoved() {
+ String catalogUuid = "catalog-1";
+ IcebergConfig config =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ catalogUuid));
+ InMemoryCatalog original =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+ Namespace namespace = Namespace.of("removed");
+ original.createNamespace(namespace);
+
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ InMemoryCatalog replacement =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ Assertions.assertNotSame(original, replacement);
+ Assertions.assertFalse(replacement.namespaceExists(namespace));
+ }
+
@Test
void testLoadCatalog() {
Catalog catalog;
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
index ce2df6910a..3a3c518f30 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
@@ -30,6 +30,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
@@ -84,6 +85,9 @@ public class IcebergCatalogWrapperManager implements
AutoCloseable {
ident -> {
if (ident.namespace().level(0).equals(metalakeName)) {
catalogWrapperCache.invalidate(ident.name());
+ if
(ident.name().equals(configProvider.getDefaultCatalogName())) {
+
catalogWrapperCache.invalidate(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG);
+ }
}
});
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
index 43ec66812e..5195502b3f 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
@@ -116,7 +116,9 @@ public class DynamicIcebergConfigProvider implements
IcebergConfigProvider {
// fields into the properties map so the JDBC backend can connect.
Map<String, String> catalogProperties;
if (catalog instanceof BaseCatalog) {
- catalogProperties = ((BaseCatalog<?>)
catalog).propertiesWithCredentialProviders();
+ BaseCatalog<?> baseCatalog = (BaseCatalog<?>) catalog;
+ catalogProperties = new
HashMap<>(baseCatalog.propertiesWithCredentialProviders());
+ catalogProperties.put(IcebergConstants.CATALOG_UUID,
baseCatalog.entity().id().toString());
} else {
catalogProperties = new HashMap<>(catalog.properties());
if (catalog instanceof SupportsCredentials) {
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
index 2d4386b730..45eb62f69e 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
@@ -21,9 +21,13 @@ package org.apache.gravitino.iceberg.service;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
@@ -42,11 +46,12 @@ import org.mockito.Mockito;
public class TestIcebergCatalogWrapperManagerForREST {
private static final String DEFAULT_CATALOG = "memory";
+ private static CatalogManager mockCatalogManager;
@BeforeAll
public static void setup() throws IllegalAccessException {
// Mock CatalogManager for GravitinoEnv to avoid initialization errors
- CatalogManager mockCatalogManager = Mockito.mock(CatalogManager.class);
+ mockCatalogManager = Mockito.mock(CatalogManager.class);
FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager",
mockCatalogManager, true);
}
@@ -145,6 +150,45 @@ public class TestIcebergCatalogWrapperManagerForREST {
Assertions.assertEquals(CatalogWrapperForREST.class, wrapper.getClass());
}
+ @Test
+ public void testDefaultCatalogAliasInvalidatedWhenCatalogRemoved() throws
Exception {
+ Mockito.clearInvocations(mockCatalogManager);
+ AtomicReference<Consumer<NameIdentifier>> removeListener = new
AtomicReference<>();
+ Mockito.doAnswer(
+ invocation -> {
+ removeListener.set(invocation.getArgument(0));
+ return null;
+ })
+ .when(mockCatalogManager)
+ .addCatalogCacheRemoveListener(Mockito.any());
+
+ IcebergConfigProvider configProvider =
Mockito.mock(IcebergConfigProvider.class);
+ Mockito.when(configProvider.getDefaultCatalogName()).thenReturn("test");
+ IcebergConfig icebergConfig =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ Mockito.when(
+
configProvider.getIcebergCatalogConfig(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG))
+ .thenReturn(Optional.of(icebergConfig));
+
+ try (IcebergCatalogWrapperManager manager =
+ new IcebergCatalogWrapperManager(Maps.newHashMap(), configProvider,
true, "metalake")) {
+ IcebergRESTServerContext.create(configProvider, false, false, true,
manager);
+ CatalogWrapperForREST original =
+
manager.getCatalogWrapper(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG);
+
+ removeListener.get().accept(NameIdentifier.of("metalake", "test"));
+
+ CatalogWrapperForREST replacement =
+
manager.getCatalogWrapper(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG);
+ Assertions.assertNotSame(original, replacement);
+ }
+ }
+
private static IcebergCatalogWrapperManager newManager() {
Map<String, String> config = Maps.newHashMap();
IcebergConfigProvider configProvider =
IcebergConfigProviderFactory.create(config);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
index d5b2024152..b9a7b2ebfd 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
@@ -35,10 +35,12 @@ import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.CatalogDispatcher;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.connector.BaseCatalog;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.jdbc.JdbcCatalog;
@@ -584,4 +586,34 @@ public class TestDynamicIcebergConfigProvider {
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
}
+
+ @Test
+ public void testAuxiliaryCatalogConfigIncludesCatalogUuid() {
+ String metalakeName = "test_metalake";
+ String catalogName = "memory_catalog";
+ Map<String, String> catalogProperties =
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_BACKEND_NAME,
+ catalogName);
+
+ @SuppressWarnings("unchecked")
+ BaseCatalog<?> baseCatalog = Mockito.mock(BaseCatalog.class);
+ CatalogEntity catalogEntity = Mockito.mock(CatalogEntity.class);
+ Mockito.when(baseCatalog.provider()).thenReturn("lakehouse-iceberg");
+
Mockito.when(baseCatalog.propertiesWithCredentialProviders()).thenReturn(catalogProperties);
+ Mockito.when(baseCatalog.entity()).thenReturn(catalogEntity);
+ Mockito.when(catalogEntity.id()).thenReturn(9L);
+
+ DynamicIcebergConfigProvider provider = new DynamicIcebergConfigProvider();
+ provider.initialize(Map.of(IcebergConstants.GRAVITINO_METALAKE,
metalakeName));
+ setMockCatalogFetcher(provider, Map.of(catalogName, baseCatalog));
+
+ Optional<IcebergConfig> config =
provider.getIcebergCatalogConfig(catalogName);
+
+ Assertions.assertTrue(config.isPresent());
+ Assertions.assertEquals(
+ "9",
config.get().getIcebergCatalogProperties().get(IcebergConstants.CATALOG_UUID));
+ }
}
