diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java index 5310fb4321..25e7d9a247 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java @@ -30,6 +30,7 @@ import org.apache.amoro.exception.ObjectNotExistsException; import org.apache.amoro.formats.iceberg.IcebergTable; import org.apache.amoro.io.AuthenticatedFileIO; +import org.apache.amoro.io.AuthenticatedFileIOs; import org.apache.amoro.mixed.InternalMixedIcebergCatalog; import org.apache.amoro.server.AmoroManagementConf; import org.apache.amoro.server.RestCatalogService; @@ -56,6 +57,8 @@ import org.apache.iceberg.rest.RESTCatalog; import org.apache.iceberg.rest.requests.CreateTableRequest; +import java.util.Map; + public class InternalCatalogImpl extends InternalCatalog { private static final String URI = "uri"; @@ -117,23 +120,28 @@ private AmoroTable loadIcebergTable( String database, String tableName, InternalTableHandler handler) { TableMetadata tableMetadata = handler.tableMetadata(); TableOperations ops = handler.newTableOperator(); - - BaseTable table = - new BaseTable( - ops, - TableIdentifier.of( - tableMetadata.getTableIdentifier().getDatabase(), - tableMetadata.getTableIdentifier().getTableName()) - .toString()); + BaseTable table = newBaseTable(tableMetadata, ops); + CatalogMeta catalogMeta = getMetadata(); + Map catalogProperties = catalogMeta.getCatalogProperties(); + Map tableProperties = table.properties(); + AuthenticatedFileIO fileIO = (AuthenticatedFileIO) ops.io(); + if (AuthenticatedFileIOs.isHdfsImpersonationEnabledForOptimizingCommit( + tableProperties, catalogProperties)) { + handler.close(); + fileIO = InternalTableUtil.newIcebergFileIo(catalogMeta, tableProperties); + ops = new InternalIcebergHandler(tableMetadata, fileIO).newTableOperator(); + table = newBaseTable(tableMetadata, ops); + } org.apache.amoro.table.TableIdentifier tableIdentifier = org.apache.amoro.table.TableIdentifier.of(name(), database, tableName); AmoroTable amoroTable = IcebergTable.newIcebergTable( tableIdentifier, table, - CatalogUtil.buildMetaStore(getMetadata()), - getMetadata().getCatalogProperties()); - fileIOCloser.put(amoroTable, ops.io()); + fileIO, + CatalogUtil.buildMetaStore(catalogMeta).getConfiguration(), + catalogProperties); + fileIOCloser.put(amoroTable, fileIO); return amoroTable; } @@ -142,12 +150,22 @@ private AmoroTable loadMixedIcebergTable( TableMetadata tableMetadata = handler.tableMetadata(); org.apache.amoro.table.TableIdentifier tableIdentifier = org.apache.amoro.table.TableIdentifier.of(name(), database, tableName); - AuthenticatedFileIO fileIO = InternalTableUtil.newIcebergFileIo(getMetadata()); + CatalogMeta catalogMeta = getMetadata(); + Map catalogProperties = catalogMeta.getCatalogProperties(); + TableOperations baseOps = handler.newTableOperator(); + BaseTable baseTable = newBaseTable(tableMetadata, baseOps); + Map tableProperties = baseTable.properties(); + AuthenticatedFileIO fileIO = (AuthenticatedFileIO) baseOps.io(); + if (AuthenticatedFileIOs.isHdfsImpersonationEnabledForOptimizingCommit( + tableProperties, catalogProperties)) { + handler.close(); + fileIO = InternalTableUtil.newIcebergFileIo(catalogMeta, tableProperties); + baseTable = loadTableStore(tableMetadata, false, fileIO); + } MixedTable mixedIcebergTable; - BaseTable baseTable = loadTableStore(tableMetadata, false); if (InternalTableUtil.isKeyedMixedTable(tableMetadata)) { - BaseTable changeTable = loadTableStore(tableMetadata, true); + BaseTable changeTable = loadTableStore(tableMetadata, true, fileIO); PrimaryKeySpec.Builder keySpecBuilder = PrimaryKeySpec.builderFor(baseTable.schema()); tableMetadata.buildTableMeta().getKeySpec().getFields().forEach(keySpecBuilder::addColumn); @@ -158,13 +176,12 @@ private AmoroTable loadMixedIcebergTable( tableMetadata.getTableLocation(), keySpec, new BasicKeyedTable.BaseInternalTable( - tableIdentifier, baseTable, fileIO, getMetadata().getCatalogProperties()), + tableIdentifier, baseTable, fileIO, catalogProperties), new BasicKeyedTable.ChangeInternalTable( - tableIdentifier, changeTable, fileIO, getMetadata().getCatalogProperties())); + tableIdentifier, changeTable, fileIO, catalogProperties)); } else { mixedIcebergTable = - new BasicUnkeyedTable( - tableIdentifier, baseTable, fileIO, getMetadata().getCatalogProperties()); + new BasicUnkeyedTable(tableIdentifier, baseTable, fileIO, catalogProperties); } AmoroTable amoroTable = new org.apache.amoro.formats.mixed.MixedTable(mixedIcebergTable, TableFormat.MIXED_ICEBERG); @@ -172,8 +189,14 @@ tableIdentifier, baseTable, fileIO, getMetadata().getCatalogProperties()), return amoroTable; } - private BaseTable loadTableStore(TableMetadata tableMetadata, boolean isChangeStore) { - TableOperations ops = newTableStoreHandler(tableMetadata, isChangeStore).newTableOperator(); + private BaseTable loadTableStore( + TableMetadata tableMetadata, boolean isChangeStore, AuthenticatedFileIO fileIO) { + TableOperations ops = + newTableStoreHandler(tableMetadata, isChangeStore, fileIO).newTableOperator(); + return newBaseTable(tableMetadata, ops); + } + + private BaseTable newBaseTable(TableMetadata tableMetadata, TableOperations ops) { return new BaseTable( ops, TableIdentifier.of( @@ -248,6 +271,11 @@ private InternalTableHandler newTableStoreHandler( return new InternalMixedIcebergHandler(getMetadata(), metadata, isChangeStore); } + private InternalTableHandler newTableStoreHandler( + TableMetadata metadata, boolean isChangeStore, AuthenticatedFileIO fileIO) { + return new InternalMixedIcebergHandler(getMetadata(), metadata, isChangeStore, fileIO); + } + private Cache, FileIO> newFileIOCloser() { return Caffeine.newBuilder() .weakKeys() diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java index c090979a27..cfdf8c2e7c 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java @@ -27,6 +27,7 @@ import org.apache.amoro.exception.OptimizingClosedException; import org.apache.amoro.exception.PersistenceException; import org.apache.amoro.exception.TaskNotFoundException; +import org.apache.amoro.io.AuthenticatedFileIOs; import org.apache.amoro.optimizing.MetricsSummary; import org.apache.amoro.optimizing.OptimizingType; import org.apache.amoro.optimizing.RewriteFilesInput; @@ -923,10 +924,12 @@ public MetricsSummary getSummary() { private UnKeyedTableCommit buildCommit() { MixedTable table = - (MixedTable) - catalogManager - .loadTable(tableRuntime.getTableIdentifier().getIdentifier()) - .originalTable(); + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + (MixedTable) + catalogManager + .loadTable(tableRuntime.getTableIdentifier().getIdentifier()) + .originalTable()); if (table.isUnkeyedTable()) { return new UnKeyedTableCommit(targetSnapshotId, table, taskMap.values()); } else { diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergHandler.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergHandler.java index 85a9e201b1..99647924f2 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergHandler.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergHandler.java @@ -37,8 +37,12 @@ public class InternalIcebergHandler implements InternalTableHandler tableProperties) { Map catalogProperties = meta.getCatalogProperties(); TableMetaStore store = CatalogUtil.buildMetaStore(meta); Configuration conf = store.getConfiguration(); @@ -95,7 +107,11 @@ public static AuthenticatedFileIO newIcebergFileIo(CatalogMeta meta) { } String ioImpl = catalogProperties.getOrDefault(CatalogProperties.FILE_IO_IMPL, defaultImpl); FileIO fileIO = org.apache.iceberg.CatalogUtil.loadFileIO(ioImpl, catalogProperties, conf); - return AuthenticatedFileIOs.buildAdaptIcebergFileIO(store, fileIO); + if (tableProperties == null) { + return AuthenticatedFileIOs.buildAdaptIcebergFileIO(store, fileIO); + } + return AuthenticatedFileIOs.buildAdaptIcebergFileIO( + store, fileIO, tableProperties, catalogProperties); } /** diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/internal/TestInternalIcebergHandlerImpersonation.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/internal/TestInternalIcebergHandlerImpersonation.java new file mode 100644 index 0000000000..c03a5037b6 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/internal/TestInternalIcebergHandlerImpersonation.java @@ -0,0 +1,87 @@ +/* + * 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.amoro.server.table.internal; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; + +import org.apache.amoro.api.CatalogMeta; +import org.apache.amoro.io.AuthenticatedFileIO; +import org.apache.amoro.io.AuthenticatedFileIOs; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.server.table.TableMetadata; +import org.apache.amoro.server.utils.InternalTableUtil; +import org.apache.amoro.table.TableProperties; +import org.apache.hadoop.security.UserGroupInformation; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.HashMap; +import java.util.Map; + +public class TestInternalIcebergHandlerImpersonation { + + @Test + public void testInternalHandlerUsesTableOwner() { + String catalogUser = "amoro-service"; + String tableOwner = "table-owner"; + TableMetadata tableMetadata = mock(TableMetadata.class); + CatalogMeta catalogMeta = newCatalogMeta(catalogUser); + AuthenticatedFileIO fileIO = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + InternalTableUtil.newIcebergFileIo( + catalogMeta, Map.of(TableProperties.OWNER, tableOwner))); + InternalIcebergHandler handler = new InternalIcebergHandler(tableMetadata, fileIO); + try { + AuthenticatedFileIO handlerFileIO = (AuthenticatedFileIO) handler.newTableOperator().io(); + assertEquals(tableOwner, handlerFileIO.doAs(this::currentUser)); + } finally { + handler.close(); + } + } + + private static CatalogMeta newCatalogMeta(String catalogUser) { + String emptyConfiguration = + Base64.getEncoder().encodeToString("".getBytes(StandardCharsets.UTF_8)); + Map storageConfigs = new HashMap<>(); + storageConfigs.put( + CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE, + CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HADOOP); + storageConfigs.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE, emptyConfiguration); + storageConfigs.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE, emptyConfiguration); + + Map authConfigs = new HashMap<>(); + authConfigs.put( + CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE, + CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE); + authConfigs.put(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME, catalogUser); + + Map catalogProperties = new HashMap<>(); + catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, "file:///tmp/amoro"); + catalogProperties.put(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + return new CatalogMeta("test", "ams", storageConfigs, authConfigs, catalogProperties); + } + + private String currentUser() throws IOException { + return UserGroupInformation.getCurrentUser().getShortUserName(); + } +} diff --git a/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java b/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java index 10f055b80f..ef1a70cb7b 100644 --- a/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java +++ b/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java @@ -98,6 +98,8 @@ public class CatalogMetaProperties { public static final String DATA_EXPIRATION_PREFIX = "data-expire."; public static final String TABLE_TRASH_PREFIX = "table-trash."; public static final String AUTO_CREATE_TAG_PREFIX = "tag.auto-create."; + public static final String HDFS_IMPERSONATION_PREFIX = "hdfs.impersonation."; + public static final String HDFS_IMPERSONATION_ENABLED = HDFS_IMPERSONATION_PREFIX + "enabled"; // mixed-format properties public static final String MIXED_FORMAT_TABLE_STORE_SEPARATOR = diff --git a/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java b/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java index f74c32b612..bc86d41cf7 100644 --- a/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java +++ b/amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java @@ -231,6 +231,12 @@ public boolean isKerberosAuthMethod() { return AUTH_METHOD_KERBEROS.equalsIgnoreCase(authMethod); } + /** Returns whether this meta store uses an authentication mode that supports proxy users. */ + public boolean supportsHadoopImpersonation() { + return AUTH_METHOD_SIMPLE.equalsIgnoreCase(authMethod) + || AUTH_METHOD_KERBEROS.equalsIgnoreCase(authMethod); + } + public String getHadoopUsername() { return hadoopUsername; } diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalog.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalog.java index fbe1ff3731..e6e4b0c5e1 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalog.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalog.java @@ -20,6 +20,7 @@ import org.apache.amoro.AmoroTable; import org.apache.amoro.FormatCatalog; +import org.apache.amoro.io.TableOwnerResolver; import org.apache.amoro.table.TableMetaStore; import org.apache.amoro.utils.MixedFormatCatalogUtil; import org.apache.iceberg.Table; @@ -39,14 +40,24 @@ public class IcebergCatalog implements FormatCatalog { private final Catalog icebergCatalog; private final TableMetaStore metaStore; private final Map properties; + private final String metastoreType; public IcebergCatalog(Catalog catalog, Map properties, TableMetaStore metaStore) { + this(catalog, null, properties, metaStore); + } + + public IcebergCatalog( + Catalog catalog, + String metastoreType, + Map properties, + TableMetaStore metaStore) { this.icebergCatalog = MixedFormatCatalogUtil.buildCacheCatalog(catalog, properties); if (catalog instanceof SupportsNamespaces) { this.asNamespaceCatalog = (SupportsNamespaces) catalog; } this.metaStore = metaStore; this.properties = properties; + this.metastoreType = metastoreType; } @Override @@ -102,11 +113,13 @@ public AmoroTable loadTable(String database, String table) { () -> { try { Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of(database, table)); + org.apache.amoro.table.TableIdentifier identifier = + org.apache.amoro.table.TableIdentifier.of(icebergCatalog.name(), database, table); + String tableOwner = + TableOwnerResolver.resolve( + metastoreType, identifier, icebergTable, properties, metaStore); return IcebergTable.newIcebergTable( - org.apache.amoro.table.TableIdentifier.of(icebergCatalog.name(), database, table), - icebergTable, - metaStore, - properties); + identifier, icebergTable, metaStore, properties, tableOwner); } catch (NoSuchTableException e) { throw new org.apache.amoro.NoSuchTableException(e); } diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalogFactory.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalogFactory.java index 432fa81b38..683ed00b23 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalogFactory.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalogFactory.java @@ -42,7 +42,7 @@ public FormatCatalog create( Catalog icebergCatalog = CatalogUtil.buildIcebergCatalog(name, properties, metaStore.getConfiguration()); - return new IcebergCatalog(icebergCatalog, properties, metaStore); + return new IcebergCatalog(icebergCatalog, metastoreType, properties, metaStore); } @Override diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergTable.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergTable.java index 6377f37831..c122daf0da 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergTable.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergTable.java @@ -28,6 +28,7 @@ import org.apache.amoro.table.TableMetaStore; import org.apache.amoro.table.UnkeyedTable; import org.apache.amoro.utils.MixedFormatCatalogUtil; +import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; @@ -49,14 +50,38 @@ public static IcebergTable newIcebergTable( Table icebergTable, TableMetaStore metaStore, Map catalogProperties) { + return newIcebergTable( + identifier, + icebergTable, + metaStore, + catalogProperties, + icebergTable.properties().get(org.apache.amoro.table.TableProperties.OWNER)); + } + + public static IcebergTable newIcebergTable( + TableIdentifier identifier, + Table icebergTable, + TableMetaStore metaStore, + Map catalogProperties, + String tableOwner) { AuthenticatedFileIO io = - AuthenticatedFileIOs.buildAdaptIcebergFileIO(metaStore, icebergTable.io()); + AuthenticatedFileIOs.buildAdaptIcebergFileIO( + metaStore, icebergTable.io(), icebergTable.properties(), catalogProperties, tableOwner); + return newIcebergTable( + identifier, icebergTable, io, metaStore.getConfiguration(), catalogProperties); + } + public static IcebergTable newIcebergTable( + TableIdentifier identifier, + Table icebergTable, + AuthenticatedFileIO io, + Configuration configuration, + Map catalogProperties) { UnkeyedTable wrapped = new BasicUnkeyedTable( identifier, MixedFormatCatalogUtil.useMixedTableOperations( - icebergTable, icebergTable.location(), io, metaStore.getConfiguration()), + icebergTable, icebergTable.location(), io, configuration), io, catalogProperties) { @Override diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedFileIOs.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedFileIOs.java index 5317596c98..958bd77943 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedFileIOs.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedFileIOs.java @@ -18,19 +18,24 @@ package org.apache.amoro.io; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions; import org.apache.amoro.table.TableIdentifier; import org.apache.amoro.table.TableMetaStore; import org.apache.amoro.table.TableProperties; import org.apache.amoro.utils.MixedFormatCatalogUtil; +import org.apache.commons.lang3.StringUtils; import org.apache.iceberg.hadoop.HadoopFileIO; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.util.PropertyUtil; import java.util.Map; +import java.util.function.Supplier; public class AuthenticatedFileIOs { public static final boolean CLOSE_TRASH = true; + private static final ThreadLocal OPTIMIZING_COMMIT_IMPERSONATION = new ThreadLocal<>(); public static AuthenticatedHadoopFileIO buildRecoverableHadoopFileIO( TableIdentifier tableIdentifier, @@ -38,25 +43,44 @@ public static AuthenticatedHadoopFileIO buildRecoverableHadoopFileIO( Map tableProperties, TableMetaStore tableMetaStore, Map catalogProperties) { - tableProperties = + return buildRecoverableHadoopFileIO( + tableIdentifier, + tableLocation, + tableProperties, + tableMetaStore, + catalogProperties, + tableProperties.get(TableProperties.OWNER)); + } + + public static AuthenticatedHadoopFileIO buildRecoverableHadoopFileIO( + TableIdentifier tableIdentifier, + String tableLocation, + Map tableProperties, + TableMetaStore tableMetaStore, + Map catalogProperties, + String tableOwner) { + String proxyUser = + resolveProxyUser(tableProperties, catalogProperties, tableMetaStore, tableOwner); + Map effectiveTableProperties = MixedFormatCatalogUtil.mergeCatalogPropertiesToTable(tableProperties, catalogProperties); if (!CLOSE_TRASH && PropertyUtil.propertyAsBoolean( - tableProperties, + effectiveTableProperties, TableProperties.ENABLE_TABLE_TRASH, TableProperties.ENABLE_TABLE_TRASH_DEFAULT)) { - AuthenticatedHadoopFileIO fileIO = new AuthenticatedHadoopFileIO(tableMetaStore); + AuthenticatedHadoopFileIO fileIO = new AuthenticatedHadoopFileIO(tableMetaStore, proxyUser); TableTrashManager trashManager = - TableTrashManagers.build(tableIdentifier, tableLocation, tableProperties, fileIO); + TableTrashManagers.build( + tableIdentifier, tableLocation, effectiveTableProperties, fileIO); String trashFilePattern = PropertyUtil.propertyAsString( - tableProperties, + effectiveTableProperties, TableProperties.TABLE_TRASH_FILE_PATTERN, TableProperties.TABLE_TRASH_FILE_PATTERN_DEFAULT); - return new RecoverableHadoopFileIO(tableMetaStore, trashManager, trashFilePattern); + return new RecoverableHadoopFileIO(tableMetaStore, trashManager, trashFilePattern, proxyUser); } else { - return new AuthenticatedHadoopFileIO(tableMetaStore); + return new AuthenticatedHadoopFileIO(tableMetaStore, proxyUser); } } @@ -66,10 +90,102 @@ public static AuthenticatedHadoopFileIO buildHadoopFileIO(TableMetaStore tableMe public static AuthenticatedFileIO buildAdaptIcebergFileIO( TableMetaStore tableMetaStore, FileIO io) { + return buildAdaptIcebergFileIOWithProxyUser(tableMetaStore, io, null); + } + + public static AuthenticatedFileIO buildAdaptIcebergFileIO( + TableMetaStore tableMetaStore, + FileIO io, + Map tableProperties, + Map catalogProperties) { + String tableOwner = tableProperties == null ? null : tableProperties.get(TableProperties.OWNER); + return buildAdaptIcebergFileIO( + tableMetaStore, io, tableProperties, catalogProperties, tableOwner); + } + + public static AuthenticatedFileIO buildAdaptIcebergFileIO( + TableMetaStore tableMetaStore, + FileIO io, + Map tableProperties, + Map catalogProperties, + String tableOwner) { + return buildAdaptIcebergFileIOWithProxyUser( + tableMetaStore, + io, + resolveProxyUser(tableProperties, catalogProperties, tableMetaStore, tableOwner)); + } + + private static AuthenticatedFileIO buildAdaptIcebergFileIOWithProxyUser( + TableMetaStore tableMetaStore, FileIO io, String proxyUser) { if (io instanceof HadoopFileIO) { - return buildHadoopFileIO(tableMetaStore); + return new AuthenticatedHadoopFileIO(tableMetaStore, proxyUser); } else { + Preconditions.checkArgument( + proxyUser == null, + "HDFS impersonation requires HadoopFileIO, but the table uses %s", + io.getClass().getName()); return new AuthenticatedFileIOAdapter(io); } } + + /** + * Runs a table loader in the optimizing-commit scope. FileIOs created during the load retain the + * resolved table owner after the scope is restored. + */ + public static T withOptimizingCommitImpersonation(Supplier tableLoader) { + Boolean previous = OPTIMIZING_COMMIT_IMPERSONATION.get(); + OPTIMIZING_COMMIT_IMPERSONATION.set(true); + try { + return tableLoader.get(); + } finally { + if (previous == null) { + OPTIMIZING_COMMIT_IMPERSONATION.remove(); + } else { + OPTIMIZING_COMMIT_IMPERSONATION.set(previous); + } + } + } + + static boolean isOptimizingCommitImpersonationActive() { + return Boolean.TRUE.equals(OPTIMIZING_COMMIT_IMPERSONATION.get()); + } + + /** Returns whether HDFS impersonation is enabled for the current optimizing-commit table load. */ + public static boolean isHdfsImpersonationEnabledForOptimizingCommit( + Map tableProperties, Map catalogProperties) { + if (!isOptimizingCommitImpersonationActive()) { + return false; + } + if (tableProperties != null + && tableProperties.containsKey(TableProperties.HDFS_IMPERSONATION_ENABLED)) { + return Boolean.parseBoolean(tableProperties.get(TableProperties.HDFS_IMPERSONATION_ENABLED)); + } + if (catalogProperties != null + && catalogProperties.containsKey(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED)) { + return Boolean.parseBoolean( + catalogProperties.get(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED)); + } + return catalogProperties != null + && Boolean.parseBoolean( + catalogProperties.get( + CatalogMetaProperties.TABLE_PROPERTIES_PREFIX + + TableProperties.HDFS_IMPERSONATION_ENABLED)); + } + + private static String resolveProxyUser( + Map tableProperties, + Map catalogProperties, + TableMetaStore tableMetaStore, + String tableOwner) { + if (!isHdfsImpersonationEnabledForOptimizingCommit(tableProperties, catalogProperties)) { + return null; + } + Preconditions.checkArgument( + tableMetaStore.supportsHadoopImpersonation(), + "HDFS impersonation requires SIMPLE or KERBEROS catalog authentication"); + Preconditions.checkArgument( + StringUtils.isNotBlank(tableOwner), + "HDFS impersonation is enabled, but the table owner is missing"); + return tableOwner; + } } diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedHadoopFileIO.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedHadoopFileIO.java index face2daaf8..4ba5b79c11 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedHadoopFileIO.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/AuthenticatedHadoopFileIO.java @@ -47,30 +47,36 @@ public class AuthenticatedHadoopFileIO extends HadoopFileIO implements AuthenticatedFileIO, SupportsPrefixOperations, SupportsFileSystemOperations { private final TableMetaStore tableMetaStore; + private final String proxyUser; AuthenticatedHadoopFileIO(TableMetaStore tableMetaStore) { + this(tableMetaStore, null); + } + + AuthenticatedHadoopFileIO(TableMetaStore tableMetaStore, String proxyUser) { super(tableMetaStore.getConfiguration()); this.tableMetaStore = tableMetaStore; + this.proxyUser = proxyUser; } @Override public InputFile newInputFile(String path) { - return tableMetaStore.doAs(() -> super.newInputFile(path)); + return doAs(() -> super.newInputFile(path)); } @Override public InputFile newInputFile(String path, long length) { - return tableMetaStore.doAs(() -> super.newInputFile(path, length)); + return doAs(() -> super.newInputFile(path, length)); } @Override public OutputFile newOutputFile(String path) { - return tableMetaStore.doAs(() -> super.newOutputFile(path)); + return doAs(() -> super.newOutputFile(path)); } @Override public void deleteFile(String path) { - tableMetaStore.doAs( + doAs( () -> { Path toDelete = new Path(path); FileSystem fs = getFs(toDelete); @@ -86,7 +92,7 @@ public void deleteFile(String path) { @Override public Iterable listDirectory(String location) { - return tableMetaStore.doAs( + return doAs( () -> { Path path = new Path(location); FileSystem fs = getFs(path); @@ -123,7 +129,7 @@ public List listWithoutDoAs(String location) { @Override public void makeDirectories(String path) { - tableMetaStore.doAs( + doAs( () -> { Path filePath = new Path(path); FileSystem fs = getFs(filePath); @@ -143,7 +149,7 @@ public void makeDirectories(String path) { @Override public boolean isDirectory(String location) { - return tableMetaStore.doAs( + return doAs( () -> { Path path = new Path(location); FileSystem fs = getFs(path); @@ -160,7 +166,7 @@ public boolean isDirectory(String location) { public boolean isEmptyDirectory(String location) { Preconditions.checkArgument( isDirectory(location), "the target location %s is not a directory", location); - return tableMetaStore.doAs( + return doAs( () -> { Path path = new Path(location); FileSystem fs = getFs(path); @@ -175,7 +181,7 @@ public boolean isEmptyDirectory(String location) { @Override public void rename(String src, String dts) { - tableMetaStore.doAs( + doAs( () -> { Path srcPath = new Path(src); Path dtsPath = new Path(dts); @@ -198,12 +204,15 @@ public void rename(String src, String dts) { @Override public T doAs(Callable callable) { - return tableMetaStore.doAs(callable); + if (proxyUser == null) { + return tableMetaStore.doAs(callable); + } + return tableMetaStore.doAsImpersonating(proxyUser, callable); } @Override public boolean exists(String path) { - return tableMetaStore.doAs( + return doAs( () -> { Path filePath = new Path(path); FileSystem fs = getFs(filePath); @@ -217,12 +226,12 @@ public boolean exists(String path) { @Override public Iterable listPrefix(String prefix) { - return tableMetaStore.doAs(() -> super.listPrefix(prefix)); + return doAs(() -> super.listPrefix(prefix)); } @Override public void deletePrefix(String prefix) { - tableMetaStore.doAs( + doAs( () -> { Path prefixToDelete = new Path(prefix); FileSystem fs = getFs(prefixToDelete); @@ -236,7 +245,7 @@ public void deletePrefix(String prefix) { @Override public void deleteFiles(Iterable pathsToDelete) throws BulkDeletionFailureException { - tableMetaStore.doAs( + doAs( () -> { super.deleteFiles(pathsToDelete); return null; diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/RecoverableHadoopFileIO.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/RecoverableHadoopFileIO.java index 5b26d88ac2..7d843df40f 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/RecoverableHadoopFileIO.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/RecoverableHadoopFileIO.java @@ -42,7 +42,15 @@ public class RecoverableHadoopFileIO extends AuthenticatedHadoopFileIO RecoverableHadoopFileIO( TableMetaStore tableMetaStore, TableTrashManager trashManager, String trashFilePattern) { - super(tableMetaStore); + this(tableMetaStore, trashManager, trashFilePattern, null); + } + + RecoverableHadoopFileIO( + TableMetaStore tableMetaStore, + TableTrashManager trashManager, + String trashFilePattern, + String proxyUser) { + super(tableMetaStore, proxyUser); this.trashManager = trashManager; this.trashFilePattern = trashFilePattern; this.pattern = diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/TableOwnerResolver.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/TableOwnerResolver.java new file mode 100644 index 0000000000..c549cc95d5 --- /dev/null +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/TableOwnerResolver.java @@ -0,0 +1,108 @@ +/* + * 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.amoro.io; + +import org.apache.amoro.hive.AuthenticatedHiveClientPool; +import org.apache.amoro.hive.HMSClient; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; +import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableMetaStore; +import org.apache.amoro.table.TableProperties; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.Table; +import org.apache.thrift.TException; + +import java.util.Map; +import java.util.function.Supplier; + +public class TableOwnerResolver { + + public static String resolve( + String metastoreType, + TableIdentifier tableIdentifier, + Table icebergTable, + Map catalogProperties, + TableMetaStore tableMetaStore) { + if (!AuthenticatedFileIOs.isOptimizingCommitImpersonationActive()) { + return null; + } + // Refresh before evaluating the table-level setting because it overrides the catalog setting + // and both the setting and owner may have changed on a cached Iceberg table. + tableMetaStore.doAs( + () -> { + icebergTable.refresh(); + return null; + }); + return resolve( + metastoreType, + tableIdentifier, + icebergTable.properties(), + catalogProperties, + () -> loadHiveOwner(tableIdentifier, catalogProperties, tableMetaStore)); + } + + @VisibleForTesting + static String resolve( + String metastoreType, + TableIdentifier tableIdentifier, + Map tableProperties, + Map catalogProperties, + Supplier hiveOwnerLoader) { + if (!AuthenticatedFileIOs.isHdfsImpersonationEnabledForOptimizingCommit( + tableProperties, catalogProperties)) { + return null; + } + if (CatalogMetaProperties.CATALOG_TYPE_HIVE.equalsIgnoreCase(metastoreType)) { + return hiveOwnerLoader.get(); + } + return tableProperties == null ? null : tableProperties.get(TableProperties.OWNER); + } + + private static String loadHiveOwner( + TableIdentifier tableIdentifier, + Map catalogProperties, + TableMetaStore tableMetaStore) { + HiveConf hiveConf = new HiveConf(tableMetaStore.getConfiguration(), TableOwnerResolver.class); + tableMetaStore.getHiveSiteLocation().ifPresent(hiveConf::addResource); + String metastoreUri = catalogProperties.get(CatalogProperties.URI); + if (StringUtils.isNotBlank(metastoreUri)) { + hiveConf.setVar(HiveConf.ConfVars.METASTOREURIS, metastoreUri); + } + + return tableMetaStore.doAs( + () -> { + HMSClient client = AuthenticatedHiveClientPool.createHiveMetaStoreClient(hiveConf); + try { + return client + .getTable(tableIdentifier.getDatabase(), tableIdentifier.getTableName()) + .getOwner(); + } catch (TException e) { + throw new IllegalStateException( + "Failed to load Hive owner for table " + tableIdentifier, e); + } finally { + client.close(); + } + }); + } + + private TableOwnerResolver() {} +} diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/BasicMixedIcebergCatalog.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/BasicMixedIcebergCatalog.java index 787ff5a2bf..21e5ee3a9e 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/BasicMixedIcebergCatalog.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/BasicMixedIcebergCatalog.java @@ -23,6 +23,7 @@ import org.apache.amoro.AmsClient; import org.apache.amoro.PooledAmsClient; import org.apache.amoro.io.AuthenticatedFileIO; +import org.apache.amoro.io.TableOwnerResolver; import org.apache.amoro.io.TableTrashManagers; import org.apache.amoro.op.CreateTableTransaction; import org.apache.amoro.properties.CatalogMetaProperties; @@ -66,6 +67,7 @@ public class BasicMixedIcebergCatalog implements MixedFormatCatalog { private MixedTables tables; private SupportsNamespaces asNamespaceCatalog; private String separator; + private String metastoreType; @Override public String name() { @@ -86,6 +88,7 @@ public void initialize(String name, Map properties, TableMetaSto org.apache.iceberg.catalog.Catalog catalog = buildIcebergCatalog(name, icebergCatalogProperties, metaStore.getConfiguration()); this.name = name; + this.metastoreType = metastoreType; this.tableMetaStore = metaStore; this.icebergCatalog = MixedFormatCatalogUtil.buildCacheCatalog(catalog, icebergCatalogProperties); @@ -160,7 +163,10 @@ public MixedTable loadTable(TableIdentifier tableIdentifier) { if (!tables.isBaseStore(base)) { throw new NoSuchTableException("table " + base.name() + " is not a mixed iceberg table"); } - return tables.loadTable(base, tableIdentifier); + String tableOwner = + TableOwnerResolver.resolve( + metastoreType, tableIdentifier, base, catalogProperties, tableMetaStore); + return tables.loadTable(base, tableIdentifier, tableOwner); } @Override diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/MixedTables.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/MixedTables.java index 0ab1553f83..84291ccbee 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/MixedTables.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/MixedTables.java @@ -103,8 +103,15 @@ protected TableIdentifier generateChangeStoreIdentifier(TableIdentifier baseIden * @return mixed format table instance. */ public MixedTable loadTable(Table base, org.apache.amoro.table.TableIdentifier tableIdentifier) { + return loadTable( + base, tableIdentifier, base.properties().get(org.apache.amoro.table.TableProperties.OWNER)); + } + + public MixedTable loadTable( + Table base, org.apache.amoro.table.TableIdentifier tableIdentifier, String tableOwner) { AuthenticatedFileIO io = - AuthenticatedFileIOs.buildAdaptIcebergFileIO(this.tableMetaStore, base.io()); + AuthenticatedFileIOs.buildAdaptIcebergFileIO( + this.tableMetaStore, base.io(), base.properties(), catalogProperties, tableOwner); PrimaryKeySpec keySpec = TablePropertyUtil.parsePrimaryKeySpec(base.schema(), base.properties()); if (!keySpec.primaryKeyExisted()) { diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java index daebf2adba..60b1d2a6de 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java @@ -357,6 +357,10 @@ private TableProperties() {} public static final String OWNER = "owner"; + public static final String HDFS_IMPERSONATION_ENABLED = + CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED; + public static final boolean HDFS_IMPERSONATION_ENABLED_DEFAULT = false; + /** table format related properties */ public static final String TABLE_FORMAT = "table-format"; @@ -407,6 +411,7 @@ private TableProperties() {} CatalogMetaProperties.DATA_EXPIRATION_PREFIX, CatalogMetaProperties.TABLE_TRASH_PREFIX, CatalogMetaProperties.AUTO_CREATE_TAG_PREFIX, + CatalogMetaProperties.HDFS_IMPERSONATION_PREFIX, // mixed format reading config keys TableProperties.SPLIT_OPEN_FILE_COST, TableProperties.SPLIT_LOOKBACK, diff --git a/amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestAuthenticatedHadoopFileIO.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestAuthenticatedHadoopFileIO.java new file mode 100644 index 0000000000..68acc3d600 --- /dev/null +++ b/amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestAuthenticatedHadoopFileIO.java @@ -0,0 +1,261 @@ +/* + * 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.amoro.io; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.table.TableMetaStore; +import org.apache.amoro.table.TableProperties; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.iceberg.hadoop.HadoopFileIO; +import org.apache.iceberg.io.FileIO; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +public class TestAuthenticatedHadoopFileIO { + + private static final String CATALOG_USER = "amoro-service"; + private static final String TABLE_OWNER = "table-owner"; + + @Test + public void testUsesCatalogUserByDefault() { + AuthenticatedFileIO fileIO = + AuthenticatedFileIOs.buildAdaptIcebergFileIO( + newSimpleMetaStore(), new HadoopFileIO(new Configuration())); + + assertEquals(CATALOG_USER, fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + } + + @Test + public void testUsesTableOwnerOnCallerAndExecutorThreads() throws Exception { + AuthenticatedFileIO fileIO = + buildForOptimizingCommit( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true")); + + assertEquals(TABLE_OWNER, fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + assertEquals( + TABLE_OWNER, + executor.submit(() -> fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)).get()); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testTableSettingOverridesCatalogSetting() { + AuthenticatedFileIO fileIO = + buildForOptimizingCommit( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.HDFS_IMPERSONATION_ENABLED, "false"), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true")); + + assertEquals(CATALOG_USER, fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + } + + @Test + public void testTableSettingEnablesWhenCatalogDisabled() { + AuthenticatedFileIO fileIO = + buildForOptimizingCommit( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of( + TableProperties.OWNER, + TABLE_OWNER, + TableProperties.HDFS_IMPERSONATION_ENABLED, + "true"), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "false")); + + assertEquals(TABLE_OWNER, fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + } + + @Test + public void testCatalogTableDefaultEnablesImpersonation() { + AuthenticatedFileIO fileIO = + buildForOptimizingCommit( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of( + CatalogMetaProperties.TABLE_PROPERTIES_PREFIX + + TableProperties.HDFS_IMPERSONATION_ENABLED, + "true")); + + assertEquals(TABLE_OWNER, fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + } + + @Test + public void testRejectsMissingOwnerAndRestoresContext() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> + buildForOptimizingCommit( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of(), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"))); + + assertTrue(exception.getMessage().contains("table owner is missing")); + + AuthenticatedFileIO ordinaryFileIO = + AuthenticatedFileIOs.buildAdaptIcebergFileIO( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of(), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true")); + assertEquals(CATALOG_USER, ordinaryFileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + } + + @Test + public void testRejectsUnsupportedAuthentication() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> + buildForOptimizingCommit( + TableMetaStore.EMPTY, + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"))); + + assertTrue(exception.getMessage().contains("SIMPLE or KERBEROS")); + } + + @Test + public void testRejectsNonHadoopFileIO() { + FileIO fileIO = mock(FileIO.class); + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> + buildForOptimizingCommit( + newSimpleMetaStore(), + fileIO, + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"))); + + assertTrue(exception.getMessage().contains("requires HadoopFileIO")); + } + + @Test + public void testCatalogSettingDoesNotAffectOrdinaryTableLoads() { + AuthenticatedFileIO fileIO = + AuthenticatedFileIOs.buildAdaptIcebergFileIO( + newSimpleMetaStore(), + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true")); + + assertEquals(CATALOG_USER, fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + } + + @Test + public void testProxyFailureDoesNotFallBackToCatalogUser() { + TableMetaStore metaStore = mock(TableMetaStore.class); + when(metaStore.getConfiguration()).thenReturn(new Configuration()); + when(metaStore.supportsHadoopImpersonation()).thenReturn(true); + SecurityException failure = new SecurityException("proxy denied"); + doThrow(failure).when(metaStore).doAsImpersonating(eq(TABLE_OWNER), any()); + AuthenticatedFileIO fileIO = + buildForOptimizingCommit( + metaStore, + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true")); + + SecurityException thrown = + assertThrows( + SecurityException.class, () -> fileIO.doAs(TestAuthenticatedHadoopFileIO::currentUser)); + + assertSame(failure, thrown); + verify(metaStore).doAsImpersonating(eq(TABLE_OWNER), any()); + verify(metaStore, never()).doAs(any()); + } + + @Test + public void testFileIOEntryUsesProxyUser() { + TableMetaStore metaStore = mock(TableMetaStore.class); + when(metaStore.getConfiguration()).thenReturn(new Configuration()); + when(metaStore.supportsHadoopImpersonation()).thenReturn(true); + doAnswer(invocation -> ((Callable) invocation.getArgument(1)).call()) + .when(metaStore) + .doAsImpersonating(eq(TABLE_OWNER), any()); + AuthenticatedFileIO fileIO = + buildForOptimizingCommit( + metaStore, + new HadoopFileIO(new Configuration()), + Map.of(TableProperties.OWNER, TABLE_OWNER), + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true")); + + fileIO.newOutputFile("file:///tmp/amoro-owner-test.metadata.json"); + + verify(metaStore).doAsImpersonating(eq(TABLE_OWNER), any()); + verify(metaStore, never()).doAs(any()); + } + + private static AuthenticatedFileIO buildForOptimizingCommit( + TableMetaStore metaStore, + FileIO fileIO, + Map tableProperties, + Map catalogProperties) { + return AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + AuthenticatedFileIOs.buildAdaptIcebergFileIO( + metaStore, fileIO, tableProperties, catalogProperties)); + } + + private static TableMetaStore newSimpleMetaStore() { + return TableMetaStore.builder() + .withCoreSite(new byte[0]) + .withHdfsSite(new byte[0]) + .withSimpleAuth(CATALOG_USER) + .build(); + } + + private static String currentUser() throws IOException { + return UserGroupInformation.getCurrentUser().getShortUserName(); + } +} diff --git a/amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestTableOwnerResolver.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestTableOwnerResolver.java new file mode 100644 index 0000000000..062cce9296 --- /dev/null +++ b/amoro-format-iceberg/src/test/java/org/apache/amoro/io/TestTableOwnerResolver.java @@ -0,0 +1,193 @@ +/* + * 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.amoro.io; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableMetaStore; +import org.apache.amoro.table.TableProperties; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.iceberg.Table; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +public class TestTableOwnerResolver { + + private static final TableIdentifier TABLE_IDENTIFIER = + TableIdentifier.of("catalog", "database", "table"); + private static final String CATALOG_USER = "catalog-user"; + + @Test + public void testRefreshesSettingAndOwnerInCatalogContext() { + AtomicBoolean refreshed = new AtomicBoolean(); + AtomicReference refreshUser = new AtomicReference<>(); + Table table = mock(Table.class); + doAnswer( + ignored -> { + refreshUser.set(currentUser()); + refreshed.set(true); + return null; + }) + .when(table) + .refresh(); + when(table.properties()) + .thenAnswer( + ignored -> + Map.of( + TableProperties.OWNER, + refreshed.get() ? "current-owner" : "stale-owner", + TableProperties.HDFS_IMPERSONATION_ENABLED, + Boolean.toString(refreshed.get()))); + + String owner = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + TableOwnerResolver.resolve( + CatalogMetaProperties.CATALOG_TYPE_HADOOP, + TABLE_IDENTIFIER, + table, + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "false"), + newSimpleMetaStore())); + + assertEquals("current-owner", owner); + assertEquals(CATALOG_USER, refreshUser.get()); + verify(table).refresh(); + } + + @Test + public void testOrdinaryLoadDoesNotRefreshIcebergMetadata() { + Table table = mock(Table.class); + + String owner = + TableOwnerResolver.resolve( + CatalogMetaProperties.CATALOG_TYPE_HADOOP, + TABLE_IDENTIFIER, + table, + enabledCatalogProperties(), + TableMetaStore.EMPTY); + + assertNull(owner); + verify(table, never()).refresh(); + verify(table, never()).properties(); + } + + @Test + public void testHiveCatalogUsesHiveOwner() { + String owner = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + TableOwnerResolver.resolve( + CatalogMetaProperties.CATALOG_TYPE_HIVE, + TABLE_IDENTIFIER, + Map.of(TableProperties.OWNER, "metadata-owner"), + enabledCatalogProperties(), + () -> "hms-owner")); + + assertEquals("hms-owner", owner); + } + + @Test + public void testNonHiveCatalogUsesIcebergOwner() { + AtomicBoolean hiveOwnerLoaded = new AtomicBoolean(); + String owner = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + TableOwnerResolver.resolve( + CatalogMetaProperties.CATALOG_TYPE_HADOOP, + TABLE_IDENTIFIER, + Map.of(TableProperties.OWNER, "metadata-owner"), + enabledCatalogProperties(), + () -> { + hiveOwnerLoaded.set(true); + return "hms-owner"; + })); + + assertEquals("metadata-owner", owner); + assertFalse(hiveOwnerLoaded.get()); + } + + @Test + public void testDisabledCatalogDoesNotLoadOwner() { + AtomicBoolean hiveOwnerLoaded = new AtomicBoolean(); + String owner = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + TableOwnerResolver.resolve( + CatalogMetaProperties.CATALOG_TYPE_HIVE, + TABLE_IDENTIFIER, + Map.of(), + Map.of(), + () -> { + hiveOwnerLoaded.set(true); + return "hms-owner"; + })); + + assertNull(owner); + assertFalse(hiveOwnerLoaded.get()); + } + + @Test + public void testTableSettingDisablesHiveOwnerLoad() { + AtomicBoolean hiveOwnerLoaded = new AtomicBoolean(); + String owner = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + TableOwnerResolver.resolve( + CatalogMetaProperties.CATALOG_TYPE_HIVE, + TABLE_IDENTIFIER, + Map.of(TableProperties.HDFS_IMPERSONATION_ENABLED, "false"), + enabledCatalogProperties(), + () -> { + hiveOwnerLoaded.set(true); + return "hms-owner"; + })); + + assertNull(owner); + assertFalse(hiveOwnerLoaded.get()); + } + + private static Map enabledCatalogProperties() { + return Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + } + + private static TableMetaStore newSimpleMetaStore() { + return TableMetaStore.builder() + .withCoreSite(new byte[0]) + .withHdfsSite(new byte[0]) + .withSimpleAuth(CATALOG_USER) + .build(); + } + + private static String currentUser() throws IOException { + return UserGroupInformation.getCurrentUser().getShortUserName(); + } +} diff --git a/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveTables.java b/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveTables.java index bf07b2f37e..d9c87544a2 100644 --- a/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveTables.java +++ b/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveTables.java @@ -30,6 +30,7 @@ import org.apache.amoro.io.TableTrashManagers; import org.apache.amoro.properties.HiveTableProperties; import org.apache.amoro.properties.MetaTableProperties; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions; import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.amoro.table.ChangeTable; @@ -77,6 +78,17 @@ public MixedHiveTables(Map catalogProperties, TableMetaStore met this.hiveClientPool = new CachedHiveClientPool(getTableMetaStore(), catalogProperties); } + @VisibleForTesting + MixedHiveTables( + Map catalogProperties, + TableMetaStore metaStore, + CachedHiveClientPool hiveClientPool) { + this.tableMetaStore = metaStore; + this.catalogProperties = catalogProperties; + this.tables = new HadoopTables(tableMetaStore.getConfiguration()); + this.hiveClientPool = Preconditions.checkNotNull(hiveClientPool); + } + protected TableMetaStore getTableMetaStore() { return tableMetaStore; } @@ -104,6 +116,7 @@ protected KeyedHiveTable loadKeyedTable(TableMeta tableMeta) { String tableLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_TABLE); String baseLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_BASE); String changeLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_CHANGE); + String tableOwner = loadHiveTableOwnerIfNeeded(tableIdentifier, tableMeta.getProperties()); AuthenticatedHadoopFileIO fileIO = AuthenticatedFileIOs.buildRecoverableHadoopFileIO( @@ -111,9 +124,10 @@ protected KeyedHiveTable loadKeyedTable(TableMeta tableMeta) { tableLocation, tableMeta.getProperties(), tableMetaStore, - catalogProperties); + catalogProperties, + tableOwner); checkPrivilege(fileIO, baseLocation); - Table baseIcebergTable = tableMetaStore.doAs(() -> tables.load(baseLocation)); + Table baseIcebergTable = fileIO.doAs(() -> tables.load(baseLocation)); UnkeyedHiveTable baseTable = new KeyedHiveTable.HiveBaseInternalTable( tableIdentifier, @@ -125,7 +139,7 @@ protected KeyedHiveTable loadKeyedTable(TableMeta tableMeta) { catalogProperties, false); - Table changeIcebergTable = tableMetaStore.doAs(() -> tables.load(changeLocation)); + Table changeIcebergTable = fileIO.doAs(() -> tables.load(changeLocation)); ChangeTable changeTable = new KeyedHiveTable.HiveChangeInternalTable( tableIdentifier, @@ -156,15 +170,17 @@ protected UnkeyedHiveTable loadUnKeyedTable(TableMeta tableMeta) { TableIdentifier tableIdentifier = TableIdentifier.of(tableMeta.getTableIdentifier()); String baseLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_BASE); String tableLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_TABLE); + String tableOwner = loadHiveTableOwnerIfNeeded(tableIdentifier, tableMeta.getProperties()); AuthenticatedHadoopFileIO fileIO = AuthenticatedFileIOs.buildRecoverableHadoopFileIO( tableIdentifier, tableLocation, tableMeta.getProperties(), tableMetaStore, - catalogProperties); + catalogProperties, + tableOwner); checkPrivilege(fileIO, baseLocation); - Table table = tableMetaStore.doAs(() -> tables.load(baseLocation)); + Table table = fileIO.doAs(() -> tables.load(baseLocation)); return new UnkeyedHiveTable( tableIdentifier, MixedFormatCatalogUtil.useMixedTableOperations( @@ -555,6 +571,31 @@ private boolean allowExistedHiveTable(TableMeta tableMeta) { return Boolean.parseBoolean(allowStringValue); } + @VisibleForTesting + String loadHiveTableOwnerIfNeeded( + TableIdentifier tableIdentifier, Map tableProperties) { + if (!AuthenticatedFileIOs.isHdfsImpersonationEnabledForOptimizingCommit( + tableProperties, catalogProperties)) { + return null; + } + return loadHiveTableOwner(tableIdentifier); + } + + private String loadHiveTableOwner(TableIdentifier tableIdentifier) { + try { + return hiveClientPool.run( + client -> + client + .getTable(tableIdentifier.getDatabase(), tableIdentifier.getTableName()) + .getOwner()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Failed to load Hive owner for table " + tableIdentifier, e); + } catch (TException e) { + throw new IllegalStateException("Failed to load Hive owner for table " + tableIdentifier, e); + } + } + public MixedTable createTableByMeta( TableMeta tableMeta, Schema schema, diff --git a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveCatalog.java b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveCatalog.java index 973e8b7c2a..fcaa04ab48 100644 --- a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveCatalog.java +++ b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveCatalog.java @@ -23,15 +23,22 @@ import org.apache.amoro.BasicTableTestHelper; import org.apache.amoro.TableFormat; +import org.apache.amoro.api.CatalogMeta; +import org.apache.amoro.catalog.CatalogTestHelper; import org.apache.amoro.catalog.TestMixedCatalog; import org.apache.amoro.hive.TestHMS; +import org.apache.amoro.io.AuthenticatedFileIOs; +import org.apache.amoro.properties.CatalogMetaProperties; import org.apache.amoro.table.MixedTable; import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableProperties; +import org.apache.hadoop.security.UserGroupInformation; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Table; import org.apache.thrift.TException; import org.junit.Assert; import org.junit.ClassRule; +import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -76,6 +83,43 @@ protected void validateCreatedTable(MixedTable table, boolean withKey) throws Ex validateMixedHiveTableProperties(table.id()); } + @Test + public void testHdfsImpersonationUsesHiveOwner() throws Exception { + String ownerInHms = "hms-owner"; + getMixedFormatCatalog().createDatabase(BasicTableTestHelper.TEST_DB_NAME); + + CatalogMeta catalogMeta = + TEST_AMS.getAmsHandler().getCatalog(CatalogTestHelper.TEST_CATALOG_NAME); + TEST_AMS + .getAmsHandler() + .updateMeta(catalogMeta, CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + refreshMixedFormatCatalog(); + + createTestTableBuilder(false).withProperty(TableProperties.OWNER, "create-owner").create(); + + org.apache.hadoop.hive.metastore.api.Table hiveTable = + TEST_HMS + .getHiveClient() + .getTable( + BasicTableTestHelper.TEST_TABLE_ID.getDatabase(), + BasicTableTestHelper.TEST_TABLE_ID.getTableName()); + hiveTable.setOwner(ownerInHms); + TEST_HMS + .getHiveClient() + .alter_table( + BasicTableTestHelper.TEST_TABLE_ID.getDatabase(), + BasicTableTestHelper.TEST_TABLE_ID.getTableName(), + hiveTable); + + refreshMixedFormatCatalog(); + MixedTable loadedTable = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> getMixedFormatCatalog().loadTable(BasicTableTestHelper.TEST_TABLE_ID)); + Assert.assertEquals( + ownerInHms, + loadedTable.io().doAs(() -> UserGroupInformation.getCurrentUser().getShortUserName())); + } + @Override protected void assertIcebergTableStore( Table tableStore, boolean isBaseStore, boolean isKeyedTable) { diff --git a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveTablesImpersonation.java b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveTablesImpersonation.java new file mode 100644 index 0000000000..eb82b7e596 --- /dev/null +++ b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/catalog/TestMixedHiveTablesImpersonation.java @@ -0,0 +1,129 @@ +/* + * 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.amoro.hive.catalog; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.apache.amoro.client.ClientPool; +import org.apache.amoro.hive.CachedHiveClientPool; +import org.apache.amoro.hive.HMSClient; +import org.apache.amoro.io.AuthenticatedFileIO; +import org.apache.amoro.io.AuthenticatedFileIOs; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableMetaStore; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.thrift.TException; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +public class TestMixedHiveTablesImpersonation { + + @Test + public void testUsesOwnerLoadedFromHiveMetastore() throws Exception { + String catalogUser = "amoro-service"; + String tableOwner = "hms-owner"; + TableIdentifier identifier = TableIdentifier.of("catalog", "database", "table"); + TableMetaStore metaStore = + TableMetaStore.builder() + .withCoreSite(new byte[0]) + .withHdfsSite(new byte[0]) + .withSimpleAuth(catalogUser) + .build(); + Map catalogProperties = + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + + Table hiveTable = new Table(); + hiveTable.setOwner(tableOwner); + HMSClient hiveClient = mock(HMSClient.class); + when(hiveClient.getTable(identifier.getDatabase(), identifier.getTableName())) + .thenReturn(hiveTable); + CachedHiveClientPool hiveClientPool = mock(CachedHiveClientPool.class); + doAnswer( + invocation -> { + ClientPool.Action action = invocation.getArgument(0); + return action.run(hiveClient); + }) + .when(hiveClientPool) + .run(any()); + + MixedHiveTables tables = new MixedHiveTables(catalogProperties, metaStore, hiveClientPool); + String resolvedOwner = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> tables.loadHiveTableOwnerIfNeeded(identifier, Map.of())); + AuthenticatedFileIO fileIO = + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> + AuthenticatedFileIOs.buildRecoverableHadoopFileIO( + identifier, + "file:///tmp/table", + Map.of(), + metaStore, + catalogProperties, + resolvedOwner)); + + assertEquals(tableOwner, resolvedOwner); + assertEquals( + tableOwner, fileIO.doAs(() -> UserGroupInformation.getCurrentUser().getShortUserName())); + } + + @Test + public void testRestoresInterruptedStatusWhenOwnerLookupIsInterrupted() throws Exception { + String catalogUser = "amoro-service"; + TableIdentifier identifier = TableIdentifier.of("catalog", "database", "table"); + TableMetaStore metaStore = + TableMetaStore.builder() + .withCoreSite(new byte[0]) + .withHdfsSite(new byte[0]) + .withSimpleAuth(catalogUser) + .build(); + Map catalogProperties = + Map.of(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + CachedHiveClientPool hiveClientPool = mock(CachedHiveClientPool.class); + when(hiveClientPool.run(any())) + .thenAnswer( + invocation -> { + throw new InterruptedException("interrupted"); + }); + MixedHiveTables tables = new MixedHiveTables(catalogProperties, metaStore, hiveClientPool); + + Thread.interrupted(); + try { + IllegalStateException exception = + assertThrows( + IllegalStateException.class, + () -> + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> tables.loadHiveTableOwnerIfNeeded(identifier, Map.of()))); + + assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(exception.getCause() instanceof InterruptedException); + } finally { + Thread.interrupted(); + } + } +} diff --git a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestIcebergHiveAmoroCatalog.java b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestIcebergHiveAmoroCatalog.java index ba42904319..3975ccdad4 100644 --- a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestIcebergHiveAmoroCatalog.java +++ b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestIcebergHiveAmoroCatalog.java @@ -20,12 +20,20 @@ import org.apache.amoro.formats.AmoroCatalogTestHelper; import org.apache.amoro.formats.TestIcebergAmoroCatalog; +import org.apache.amoro.formats.iceberg.IcebergTable; import org.apache.amoro.hive.TestHMS; +import org.apache.amoro.io.AuthenticatedFileIOs; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.hadoop.security.UserGroupInformation; +import org.junit.Assert; import org.junit.ClassRule; +import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import java.io.IOException; +import java.util.HashMap; +import java.util.Map; @RunWith(Parameterized.class) public class TestIcebergHiveAmoroCatalog extends TestIcebergAmoroCatalog { @@ -38,7 +46,9 @@ public TestIcebergHiveAmoroCatalog(AmoroCatalogTestHelper amoroCatalogTestHel @Parameterized.Parameters(name = "{0}") public static Object[] parameters() { - return new Object[] {IcebergHiveCatalogTestHelper.defaultHelper()}; + Map properties = new HashMap<>(); + properties.put(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + return new Object[] {new IcebergHiveCatalogTestHelper("test_iceberg_catalog", properties)}; } @Override @@ -46,4 +56,29 @@ public void setupCatalog() throws IOException { catalogTestHelper.initHiveConf(TEST_HMS.getHiveConf()); super.setupCatalog(); } + + @Test + public void testHdfsImpersonationUsesHiveOwner() throws Exception { + String database = "owner_db"; + String table = "owner_table"; + String owner = "hms-owner"; + createDatabase(database); + createTable(database, table, new HashMap<>()); + + org.apache.hadoop.hive.metastore.api.Table hiveTable = + TEST_HMS.getHiveClient().getTable(database, table); + hiveTable.setOwner(owner); + TEST_HMS.getHiveClient().alter_table(database, table, hiveTable); + + IcebergTable loaded = + (IcebergTable) + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> amoroCatalog.loadTable(database, table)); + Assert.assertEquals( + owner, + loaded + .originalTable() + .io() + .doAs(() -> UserGroupInformation.getCurrentUser().getShortUserName())); + } } diff --git a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestMixedIcebergHiveAmoroCatalog.java b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestMixedIcebergHiveAmoroCatalog.java index f26c066aa3..8fdaddb5c2 100644 --- a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestMixedIcebergHiveAmoroCatalog.java +++ b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/amoro/hive/formats/TestMixedIcebergHiveAmoroCatalog.java @@ -21,11 +21,18 @@ import org.apache.amoro.formats.AmoroCatalogTestHelper; import org.apache.amoro.formats.TestMixedIcebergFormatCatalog; import org.apache.amoro.hive.TestHMS; +import org.apache.amoro.io.AuthenticatedFileIOs; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.hadoop.security.UserGroupInformation; +import org.junit.Assert; import org.junit.ClassRule; +import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import java.io.IOException; +import java.util.HashMap; +import java.util.Map; @RunWith(Parameterized.class) public class TestMixedIcebergHiveAmoroCatalog extends TestMixedIcebergFormatCatalog { @@ -38,7 +45,11 @@ public TestMixedIcebergHiveAmoroCatalog(AmoroCatalogTestHelper amoroCatalogTe @Parameterized.Parameters(name = "{0}") public static Object[] parameters() { - return new Object[] {MixedIcebergHiveCatalogTestHelper.defaultHelper()}; + Map properties = new HashMap<>(); + properties.put(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true"); + return new Object[] { + new MixedIcebergHiveCatalogTestHelper("test_mixed_iceberg_hive_catalog", properties) + }; } @Override @@ -46,4 +57,29 @@ public void setupCatalog() throws IOException { catalogTestHelper.initHiveConf(TEST_HMS.getHiveConf()); super.setupCatalog(); } + + @Test + public void testHdfsImpersonationUsesHiveOwner() throws Exception { + String database = "owner_db"; + String table = "owner_table"; + String owner = "hms-owner"; + createDatabase(database); + createTable(database, table, new HashMap<>()); + + org.apache.hadoop.hive.metastore.api.Table hiveTable = + TEST_HMS.getHiveClient().getTable(database, table); + hiveTable.setOwner(owner); + TEST_HMS.getHiveClient().alter_table(database, table, hiveTable); + + org.apache.amoro.formats.mixed.MixedTable loaded = + (org.apache.amoro.formats.mixed.MixedTable) + AuthenticatedFileIOs.withOptimizingCommitImpersonation( + () -> amoroCatalog.loadTable(database, table)); + Assert.assertEquals( + owner, + loaded + .originalTable() + .io() + .doAs(() -> UserGroupInformation.getCurrentUser().getShortUserName())); + } } diff --git a/docs/admin-guides/managing-catalogs.md b/docs/admin-guides/managing-catalogs.md index c29abb6719..4810f15a1c 100644 --- a/docs/admin-guides/managing-catalogs.md +++ b/docs/admin-guides/managing-catalogs.md @@ -76,6 +76,27 @@ Common properties include: ### Configure table properties If you want to add the same table properties to all tables under a catalog, you can add these table properties here on the catalog level. If you also configure this property on the table level, the property on the table will take effect. +### HDFS owner impersonation + +For Hadoop-backed Iceberg, Mixed-Iceberg, and Mixed-Hive tables, AMS can run HDFS FileIO used by optimizing commits as the table owner instead of the catalog service user. This includes file operations performed while committing and commit-time cleanup. This is disabled by default. Enable it with the catalog property `hdfs.impersonation.enabled=true`, or set the same property on an individual table. An explicit table value overrides the catalog value. + +This setting applies only to AMS-side optimizing commits. Optimizer task writes and other maintenance operations continue to use their existing identities. For Hive Metastore catalogs, AMS reads the current table owner from Hive Metastore. For other Iceberg and Mixed-Iceberg catalogs, including internal catalogs, it reads the `owner` property from the current Iceberg table metadata. The table must have a non-empty owner and the catalog must use SIMPLE or KERBEROS authentication with Hadoop FileIO. Amoro fails the commit instead of falling back to the catalog service user when these requirements are not met. + +The Hadoop cluster must authorize the catalog login user to create proxy users. Amoro does not special-case owner names, including the conventional `hdfs` superuser, so proxy-user rules must exclude identities that the catalog service must not assume. Hadoop proxy authorization failures are propagated and never cause a fallback to the catalog service user. For example, if the Kerberos principal has the short name `amoro`, configure appropriately restricted values in `core-site.xml`: + +```xml + + hadoop.proxyuser.amoro.hosts + trusted-amoro-hosts + + + hadoop.proxyuser.amoro.users + allowed-table-owners + +``` + +You can use `hadoop.proxyuser.amoro.groups` instead of, or together with, the users setting. Avoid wildcard values unless they are explicitly required by your security policy. See the [Hadoop proxy user documentation](https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/Superusers.html) for the authorization rules. HDFS and Ranger authorization still apply to the effective table owner. + ## REST Catalog When a user needs to create a Iceberg REST Catalog, they can choose **External Catalog Type**、**Custom Metastore Type**、**Iceberg Table Format**, configure properties include: **catalog-impl=org.apache.iceberg.rest.RESTCatalog**, **uri=$restCatalog_uri**.