pvary commented on code in PR #13187: URL: https://github.com/apache/iceberg/pull/13187#discussion_r2115567983
########## spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/security/IcebergHiveConnectorDelegationTokenProvider.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.iceberg.spark.security; + +import java.io.IOException; +import java.lang.reflect.UndeclaredThrowableException; +import java.security.PrivilegedExceptionAction; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.HiveMetaStoreClient; +import org.apache.hadoop.hive.metastore.IMetaStoreClient; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.security.Credentials; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.security.token.Token; +import org.apache.iceberg.spark.SparkCatalog; +import org.apache.iceberg.spark.SparkSessionCatalog; +import org.apache.spark.SparkConf; +import org.apache.spark.security.HadoopDelegationTokenProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import scala.Option; +import scala.Tuple2; + +public class IcebergHiveConnectorDelegationTokenProvider implements HadoopDelegationTokenProvider { + private static final Logger LOG = + LoggerFactory.getLogger(IcebergHiveConnectorDelegationTokenProvider.class); + + private static final String SERVICE_NAME = "iceberg_hive"; + private static final String SPARK_SQL_CATALOG_PREFIX = "spark.sql.catalog."; + private static final String CATALOG_TYPE = "hive"; + private static final String URI_KEY = ".uri"; + private static final String PRINCIPAL_KEY = ".hive.metastore.kerberos.principal"; + private static final String TYPE_KEY = ".type"; + + @Override + public String serviceName() { + return SERVICE_NAME; + } + + /** + * Builds a HiveConf object for the specified catalog by merging the provided Hadoop configuration + * with catalog-specific configurations from the Spark configuration. + * + * @param sparkConf The SparkConf object containing Spark configurations. + * @param hadoopConf The Hadoop Configuration object to be used as the base for the HiveConf. + * @param catalogName The name of the catalog. + * @return An Optional containing the constructed HiveConf if successful, or an empty Optional if + * an error occurs. + */ + Optional<HiveConf> buildHiveConf( + SparkConf sparkConf, Configuration hadoopConf, String catalogName) { + try { + HiveConf hiveConf = new HiveConf(hadoopConf, HiveConf.class); + Arrays.stream(sparkConf.getAllWithPrefix(SPARK_SQL_CATALOG_PREFIX + catalogName)) + .forEach(x -> hiveConf.set(x._1, x._2)); + return Optional.of(hiveConf); + } catch (Exception e) { + LOG.warn("Fail to create Hive Configuration for catalog {}: {}", catalogName, e.getMessage()); + return Optional.empty(); + } + } + + /** + * Extracts the names of Iceberg catalogs from the provided Spark configuration. This method + * filters the Spark configuration entries to identify those that correspond to Iceberg catalogs. + * It checks if the catalog type matches either `SparkSessionCatalog` or `SparkCatalog` and + * collects their names. + * + * @param sparkConf The SparkConf object containing Spark configurations. + * @return A Set of Strings representing the names of Iceberg catalogs. + */ + private Set<String> extractIcebergCatalogNames(SparkConf sparkConf) { + return Arrays.stream(sparkConf.getAllWithPrefix(SPARK_SQL_CATALOG_PREFIX)) + .filter( + entry -> { + String val = entry._2(); + return val.contains(SparkSessionCatalog.class.getName()) + || val.contains(SparkCatalog.class.getName()); + }) + .map(Tuple2::_1) + .collect(Collectors.toSet()); + } + + @Override + public boolean delegationTokensRequired(SparkConf sparkConf, Configuration hadoopConf) { + return !getRequireTokenCatalogs(sparkConf).isEmpty(); + } + + private Set<String> getRequireTokenCatalogs(SparkConf sparkConf) { + return extractIcebergCatalogNames(sparkConf).stream() + .filter(catalog -> checkDelegationTokensRequired(sparkConf, catalog)) + .collect(Collectors.toSet()); + } + + private boolean checkDelegationTokensRequired(SparkConf sparkConf, String catalogName) { + String metastoreUri = sparkConf.get(SPARK_SQL_CATALOG_PREFIX + catalogName + URI_KEY, ""); + String principal = sparkConf.get(SPARK_SQL_CATALOG_PREFIX + catalogName + PRINCIPAL_KEY, ""); + boolean isHiveType = + CATALOG_TYPE.equalsIgnoreCase( + sparkConf.get(SPARK_SQL_CATALOG_PREFIX + catalogName + TYPE_KEY, "")); + if (metastoreUri.isEmpty() + || principal.isEmpty() + || !isHiveType + || !UserGroupInformation.isSecurityEnabled()) { + return false; + } Review Comment: Could you please make sure that the java style guidelines are followed: https://iceberg.apache.org/contribute/#java-style-guidelines -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org For additional commands, e-mail: issues-h...@iceberg.apache.org
