This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to branch cherry-pick-30687f68-to-branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 4dc358c4d9fdb3e66e3885adb1089e765eb0d37b Author: roryqi <[email protected]> AuthorDate: Fri Sep 4 16:33:58 2026 +0800 [#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852) ### What changes were proposed in this pull request? - Expose the Iceberg backend owned by `IcebergCatalogOperations`. - Reuse that backend when the auxiliary Iceberg REST service serves a memory catalog. - Resolve the REST `default_catalog` alias to the configured Gravitino catalog. - Prevent the REST wrapper from closing the borrowed catalog. - Add regression tests covering namespace, table, and view creation for both explicit catalog names and the default catalog alias. ### Why are the changes needed? The auxiliary Iceberg REST service currently creates a second InMemoryCatalog. REST creates are written to that instance, while metadata import uses the instance owned by Gravitino CatalogManager. Because in-memory state is local to each instance, the import cannot find the newly created object and leaves Iceberg and Gravitino metadata inconsistent. Fix: #12851 ### Does this PR introduce _any_ user-facing change? Yes. Namespace, table, and view creation through the Iceberg REST API now succeeds with a memory backend, and the created objects are visible through the Gravitino metadata API. No public API or configuration property is added or removed. ### How was this patch tested? - `./gradlew spotlessApply` - `./gradlew :iceberg:iceberg-common:test --tests org.apache.gravitino.iceberg.common.ops.TestIcebergCatalogWrapper` - `./gradlew :catalogs:catalog-lakehouse-iceberg:compileJava` - `./gradlew :iceberg:iceberg-rest-server:test --tests org.apache.gravitino.iceberg.service.TestIcebergCatalogWrapperManagerForREST` - `./gradlew :iceberg:iceberg-rest-server:test --tests org.apache.gravitino.iceberg.service.TestIcebergCatalogWrapperManagerForREST --tests org.apache.gravitino.iceberg.service.dispatcher.TestIcebergNamespaceHookDispatcher --tests org.apache.gravitino.iceberg.service.dispatcher.TestIcebergTableHookDispatcher --tests org.apache.gravitino.iceberg.service.dispatcher.TestIcebergViewHookDispatcher` - `./gradlew :catalogs:catalog-lakehouse-iceberg:test --tests org.apache.gravitino.catalog.lakehouse.iceberg.integration.test.CatalogIcebergRestIT -PskipTests -PtestMode=embedded -PjdbcBackend=h2 -PskipDockerTests=false -PskipWeb=true` # Conflicts: # catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java # catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java # core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java # iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java # iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java # iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java --- .../lakehouse/iceberg/IcebergConstants.java | 3 + .../iceberg/IcebergCatalogOperations.java | 55 +++++- .../iceberg/TestIcebergCatalogOperations.java | 29 +++ .../lakehouse/iceberg/TestIcebergSchema.java | 3 +- .../apache/gravitino/catalog/CatalogManager.java | 22 +++ .../gravitino/connector/CatalogDropAware.java | 29 +++ .../gravitino/catalog/TestCatalogManager.java | 13 ++ .../iceberg/common/ops/IcebergCatalogWrapper.java | 6 +- .../iceberg/common/utils/IcebergCatalogUtil.java | 44 +++++ .../common/ops/TestIcebergCatalogWrapper.java | 27 +++ .../common/utils/TestIcebergCatalogUtil.java | 86 +++++++++ .../service/IcebergCatalogWrapperManager.java | 4 + .../provider/DynamicIcebergConfigProvider.java | 21 ++ .../TestIcebergCatalogWrapperManagerForREST.java | 46 ++++- .../provider/TestDynamicIcebergConfigProvider.java | 214 +++++++++++++++++++++ 15 files changed, 597 insertions(+), 5 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..311ccda7af 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,11 +126,13 @@ 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); +<<<<<<< HEAD AuthenticationConfig authenticationConfig = new AuthenticationConfig(resultConf); this.icebergCatalogWrapper = authenticationConfig.isKerberosAuth() && rawWrapper.getCatalog() instanceof SupportsKerberos @@ -135,6 +141,51 @@ public class IcebergCatalogOperations this.icebergCatalogWrapperHelper = new IcebergCatalogWrapperHelper(icebergCatalogWrapper.getCatalog()); this.icebergViewCatalogOperations = new IcebergViewCatalogOperations(icebergCatalogWrapper); +======= + try { + AuthenticationConfig authenticationConfig = new AuthenticationConfig(resultConf); + this.icebergCatalogWrapper = + authenticationConfig.isKerberosAuth() + && rawWrapper.getCatalog() instanceof SupportsKerberos + ? new KerberosAwareIcebergCatalogProxy(rawWrapper).getProxy(icebergConfig) + : rawWrapper; + this.icebergCatalogWrapperHelper = + new IcebergCatalogWrapperHelper(icebergCatalogWrapper.getCatalog()); + this.icebergViewCatalogOperations = new IcebergViewCatalogOperations(icebergCatalogWrapper); + } catch (NoSuchWarehouseException e) { + // A reachable server rejecting the `warehouse` selector is a user error. See issue #11943. + throw new IllegalArgumentException( + String.format( + "The 'warehouse' value '%s' could not be resolved by the Iceberg REST server. On " + + "the REST backend 'warehouse' selects a catalog by name on the remote server " + + "and is not a storage location; remove 'warehouse' to use the server's default " + + "catalog, or set it to a catalog name/identifier that the server recognizes.", + icebergConfig.get(IcebergConfig.CATALOG_WAREHOUSE)), + e); + } catch (RESTException e) { + throw handleRestException(e); + } + } + + @Override + public void onCatalogDropped() { + if (catalogUuid != null) { + IcebergCatalogUtil.removeMemoryCatalog(catalogUuid); + } + } + + // Maps an Iceberg REST-client exception into Gravitino's taxonomy + @VisibleForTesting + static RuntimeException handleRestException(RESTException e) { + // Base RESTException (server unreachable) or ServiceUnavailableException (503): the downstream + // dependency is not available. getClass() matches the base type only, so the 4xx/5xx subtypes + // are excluded and pass through unchanged. + if (e.getClass() == RESTException.class || e instanceof ServiceUnavailableException) { + return new ConnectionFailedException( + e, "The Iceberg REST backend is unavailable: %s", e.getMessage()); + } + return e; +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) } /** Closes the Iceberg catalog and releases the associated client pool. */ 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..e91b3eba65 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 @@ -29,6 +29,13 @@ import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.GravitinoRuntimeException; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; +<<<<<<< HEAD +======= +import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil; +import org.apache.iceberg.exceptions.BadRequestException; +import org.apache.iceberg.exceptions.RESTException; +import org.apache.iceberg.exceptions.ServiceUnavailableException; +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) import org.apache.iceberg.rest.responses.ListNamespacesResponse; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -95,4 +102,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..e58fea74f0 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,27 @@ public class CatalogManager implements CatalogDispatcher, Closeable { boolean deleted = store.delete(ident, EntityType.CATALOG, true); if (deleted) { markLocalMutation(ident); +<<<<<<< HEAD +======= + 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); + } + // Unmanaged: schemas removed only via store cascade — clean secrets captured above. + if (!managedStorage) { + for (Map<String, String> schemaProperties : unmanagedSchemaSecrets) { + secretManager.deleteSecretsFromProperties(schemaProperties); + } + } + secretManager.deleteSecretsFromProperties(catalogProperties); +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) } 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..9ae4afe292 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,28 @@ public class IcebergCatalogUtil { private static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogUtil.class); +<<<<<<< HEAD +======= + /** + * Column that Iceberg adds to the {@code iceberg_tables} control table in its V1 view-support + * migration (see {@code JdbcUtil} in iceberg-core). + */ + private static final String ICEBERG_TYPE_COLUMN = "iceberg_type"; + + private static final ConcurrentHashMap<String, InMemoryCatalog> MEMORY_CATALOGS = + new ConcurrentHashMap<>(); + +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) 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 +91,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..feb37ba4ca 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,6 +116,7 @@ public class DynamicIcebergConfigProvider implements IcebergConfigProvider { // fields into the properties map so the JDBC backend can connect. Map<String, String> catalogProperties; if (catalog instanceof BaseCatalog) { +<<<<<<< HEAD catalogProperties = ((BaseCatalog<?>) catalog).propertiesWithCredentialProviders(); } else { catalogProperties = new HashMap<>(catalog.properties()); @@ -131,6 +132,26 @@ public class DynamicIcebergConfigProvider implements IcebergConfigProvider { catalogProperties.putIfAbsent( IcebergConstants.GRAVITINO_JDBC_PASSWORD, jdbc.jdbcPassword()); }); +======= + BaseCatalog<?> baseCatalog = (BaseCatalog<?>) catalog; + Map<String, String> props = + new HashMap<>( + GravitinoEnv.getInstance() + .secretManager() + .toPlaintextProperties(baseCatalog.propertiesWithCredentialProviders())); + props.put(IcebergConstants.CATALOG_UUID, baseCatalog.entity().id().toString()); + return props; + } + Map<String, String> props = + new HashMap<>(catalog.properties() == null ? Map.of() : catalog.properties()); + try { + SupportsSecrets supportsSecrets = catalog.supportsSecrets(); + if (supportsSecrets != null) { + Map<String, String> secrets = supportsSecrets.getSecrets(); + if (secrets != null) { + props.putAll(secrets); + } +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) } } return Optional.of(getIcebergConfigFromCatalogProperties(catalogProperties)); 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..39c0464681 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 @@ -39,6 +39,17 @@ 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; +<<<<<<< HEAD +======= +import org.apache.gravitino.meta.CatalogEntity; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretManager; +import org.apache.gravitino.secret.SecretMaterial; +import org.apache.gravitino.secret.SecretPropertyUtils; +import org.apache.gravitino.secret.SecretProviderRegistry; +import org.apache.gravitino.secret.SupportsSecrets; +import org.apache.gravitino.secret.memory.InMemorySecretsProvider; +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.iceberg.hive.HiveCatalog; import org.apache.iceberg.jdbc.JdbcCatalog; @@ -584,4 +595,207 @@ public class TestDynamicIcebergConfigProvider { executor.shutdown(); executor.awaitTermination(5, TimeUnit.SECONDS); } +<<<<<<< HEAD +======= + + @Test + public void testMergeSecrets() { + String metalakeName = "test_metalake"; + String catalogName = "jdbc_catalog"; + + Catalog mockCatalog = Mockito.mock(Catalog.class); + SupportsSecrets supportsSecrets = Mockito.mock(SupportsSecrets.class); + Mockito.when(mockCatalog.provider()).thenReturn("lakehouse-iceberg"); + Mockito.when(mockCatalog.properties()) + .thenReturn( + new HashMap<String, String>() { + { + put(IcebergConstants.CATALOG_BACKEND, "jdbc"); + put(IcebergConstants.CATALOG_BACKEND_NAME, catalogName); + put(IcebergConstants.WAREHOUSE, "s3://bucket/wh"); + } + }); + Mockito.when(mockCatalog.supportsSecrets()).thenReturn(supportsSecrets); + Mockito.when(supportsSecrets.getSecrets()) + .thenReturn(Map.of(IcebergConstants.GRAVITINO_JDBC_PASSWORD, "secret-pwd")); + + Map<String, String> properties = new HashMap<>(); + properties.put(IcebergConstants.GRAVITINO_URI, "http://localhost:8090";); + properties.put(IcebergConstants.GRAVITINO_METALAKE, metalakeName); + + DynamicIcebergConfigProvider provider = new DynamicIcebergConfigProvider(); + provider.initialize(properties); + setMockCatalogFetcher(provider, Map.of(catalogName, mockCatalog)); + + Optional<IcebergConfig> config = provider.getIcebergCatalogConfig(catalogName); + Assertions.assertTrue(config.isPresent()); + Assertions.assertEquals( + "secret-pwd", + config.get().getIcebergCatalogProperties().get(IcebergConstants.GRAVITINO_JDBC_PASSWORD)); + } + + @Test + public void testCredsOverrideSecrets() { + String metalakeName = "test_metalake"; + String catalogName = "jdbc_catalog"; + + Catalog mockCatalog = + Mockito.mock( + Catalog.class, + Mockito.withSettings() + .extraInterfaces(SupportsCredentials.class, SupportsSecrets.class)); + SupportsSecrets supportsSecrets = (SupportsSecrets) mockCatalog; + SupportsCredentials supportsCredentials = (SupportsCredentials) mockCatalog; + + Mockito.when(mockCatalog.provider()).thenReturn("lakehouse-iceberg"); + Mockito.when(mockCatalog.properties()) + .thenReturn( + new HashMap<String, String>() { + { + put(IcebergConstants.CATALOG_BACKEND, "jdbc"); + put(IcebergConstants.CATALOG_BACKEND_NAME, catalogName); + } + }); + Mockito.when(mockCatalog.supportsSecrets()).thenReturn(supportsSecrets); + Mockito.when(supportsSecrets.getSecrets()) + .thenReturn( + Map.of( + IcebergConstants.GRAVITINO_JDBC_USER, + "from-secret", + IcebergConstants.GRAVITINO_JDBC_PASSWORD, + "secret-pwd")); + Mockito.when(supportsCredentials.getCredentials()) + .thenReturn(new Credential[] {new JdbcCredential("cred-user", "cred-pwd")}); + + Map<String, String> properties = new HashMap<>(); + properties.put(IcebergConstants.GRAVITINO_URI, "http://localhost:8090";); + properties.put(IcebergConstants.GRAVITINO_METALAKE, metalakeName); + + DynamicIcebergConfigProvider provider = new DynamicIcebergConfigProvider(); + provider.initialize(properties); + setMockCatalogFetcher(provider, Map.of(catalogName, mockCatalog)); + + Optional<IcebergConfig> config = provider.getIcebergCatalogConfig(catalogName); + Assertions.assertTrue(config.isPresent()); + Map<String, String> icebergProps = config.get().getIcebergCatalogProperties(); + Assertions.assertEquals("cred-user", icebergProps.get(IcebergConstants.GRAVITINO_JDBC_USER)); + Assertions.assertEquals("cred-pwd", icebergProps.get(IcebergConstants.GRAVITINO_JDBC_PASSWORD)); + } + + @Test + public void testMergeMemorySecrets() { + try (SecretManager sm = memorySecretManager()) { + Map<String, String> entityProps = new HashMap<>(); + entityProps.put(IcebergConstants.GRAVITINO_JDBC_USER, "root"); + List<SecretMaterial> writes = + sm.assembleSecretMaterials( + Map.of(IcebergConstants.GRAVITINO_JDBC_USER, "root"), + entityProps, + "catalog", + 3L, + Map.of( + IcebergConstants.GRAVITINO_JDBC_PASSWORD, + new SecretBinding("memory", "mem-jdbc-pwd")), + Map.of()); + sm.writeSecrets(writes); + Map<String, String> secrets = SecretPropertyUtils.buildSecrets(sm, entityProps); + + String metalakeName = "test_metalake"; + String catalogName = "jdbc_catalog"; + Catalog mockCatalog = Mockito.mock(Catalog.class); + SupportsSecrets supportsSecrets = Mockito.mock(SupportsSecrets.class); + Mockito.when(mockCatalog.provider()).thenReturn("lakehouse-iceberg"); + Mockito.when(mockCatalog.properties()) + .thenReturn( + new HashMap<String, String>() { + { + put(IcebergConstants.CATALOG_BACKEND, "jdbc"); + put(IcebergConstants.CATALOG_BACKEND_NAME, catalogName); + } + }); + Mockito.when(mockCatalog.supportsSecrets()).thenReturn(supportsSecrets); + Mockito.when(supportsSecrets.getSecrets()).thenReturn(secrets); + + Map<String, String> properties = new HashMap<>(); + properties.put(IcebergConstants.GRAVITINO_URI, "http://localhost:8090";); + properties.put(IcebergConstants.GRAVITINO_METALAKE, metalakeName); + + DynamicIcebergConfigProvider provider = new DynamicIcebergConfigProvider(); + provider.initialize(properties); + setMockCatalogFetcher(provider, Map.of(catalogName, mockCatalog)); + + Optional<IcebergConfig> config = provider.getIcebergCatalogConfig(catalogName); + Assertions.assertTrue(config.isPresent()); + Assertions.assertEquals( + "mem-jdbc-pwd", + config.get().getIcebergCatalogProperties().get(IcebergConstants.GRAVITINO_JDBC_PASSWORD)); + } + } + + @Test + public void testAuxPlaintext() throws Exception { + try (SecretManager sm = memorySecretManager()) { + String metalakeName = "test_metalake"; + String catalogName = "jdbc_catalog"; + + Map<String, String> entityProps = new HashMap<>(); + entityProps.put(IcebergConstants.CATALOG_BACKEND, "jdbc"); + entityProps.put(IcebergConstants.CATALOG_BACKEND_NAME, catalogName); + entityProps.put(IcebergConstants.GRAVITINO_JDBC_USER, "root"); + List<SecretMaterial> writes = + sm.assembleSecretMaterials( + Map.of(IcebergConstants.GRAVITINO_JDBC_USER, "root"), + entityProps, + "catalog", + 9L, + Map.of( + IcebergConstants.GRAVITINO_JDBC_PASSWORD, + new SecretBinding("memory", "aux-mem-pwd")), + Map.of()); + sm.writeSecrets(writes); + + @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(entityProps); + Mockito.when(baseCatalog.entity()).thenReturn(catalogEntity); + Mockito.when(catalogEntity.id()).thenReturn(9L); + + FieldUtils.writeField(GravitinoEnv.getInstance(), "secretManager", sm, true); + + Map<String, String> properties = new HashMap<>(); + properties.put(IcebergConstants.GRAVITINO_URI, "http://localhost:8090";); + properties.put(IcebergConstants.GRAVITINO_METALAKE, metalakeName); + + DynamicIcebergConfigProvider provider = new DynamicIcebergConfigProvider(); + provider.initialize(properties); + setMockCatalogFetcher(provider, Map.of(catalogName, baseCatalog)); + + Optional<IcebergConfig> config = provider.getIcebergCatalogConfig(catalogName); + Assertions.assertTrue(config.isPresent()); + Assertions.assertEquals( + "aux-mem-pwd", + config.get().getIcebergCatalogProperties().get(IcebergConstants.GRAVITINO_JDBC_PASSWORD)); + Assertions.assertEquals( + "root", + config.get().getIcebergCatalogProperties().get(IcebergConstants.GRAVITINO_JDBC_USER)); + Assertions.assertEquals( + "9", config.get().getIcebergCatalogProperties().get(IcebergConstants.CATALOG_UUID)); + } + } + + private static SecretManager memorySecretManager() { + Config config = new Config(false) {}; + Properties properties = new Properties(); + properties.setProperty(SecretProviderRegistry.GRAVITINO_SECRET_PROVIDERS, "memory"); + properties.setProperty( + SecretProviderRegistry.GRAVITINO_SECRET_PROVIDER_PREFIX + + "memory." + + SecretProviderRegistry.CLASS_NAME, + InMemorySecretsProvider.class.getName()); + config.loadFromProperties(properties); + return new SecretManager(config); + } +>>>>>>> 30687f68f ([#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary mode (#12852)) }
