pvary commented on code in PR #22694: URL: https://github.com/apache/flink/pull/22694#discussion_r1227217764
########## flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/security/token/HiveDelegationTokenProvider.java: ########## @@ -0,0 +1,148 @@ +/* + * 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.flink.table.security.token; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.security.token.DelegationTokenProvider; +import org.apache.flink.runtime.security.token.hadoop.HadoopDelegationTokenConverter; +import org.apache.flink.runtime.security.token.hadoop.KerberosLoginProvider; +import org.apache.flink.runtime.util.HadoopUtils; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.Preconditions; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.metadata.Hive; +import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier; +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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.security.PrivilegedExceptionAction; +import java.util.Optional; + +/** Delegation token provider for Hive. */ +@Internal +public class HiveDelegationTokenProvider implements DelegationTokenProvider { + + private static final Logger LOG = LoggerFactory.getLogger(HiveDelegationTokenProvider.class); + + org.apache.hadoop.conf.Configuration hiveConf; + + private KerberosLoginProvider kerberosLoginProvider; + + private static final Text TOKEN_ALIAS = new Text("hive.server2.delegation.token"); + + @Override + public String serviceName() { + return "HiveServer2"; + } + + @Override + public void init(Configuration configuration) throws Exception { + hiveConf = getHiveConfiguration(configuration); + kerberosLoginProvider = new KerberosLoginProvider(configuration); + } + + private org.apache.hadoop.conf.Configuration getHiveConfiguration(Configuration conf) { + try { + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopUtils.getHadoopConfiguration(conf); + hiveConf = new HiveConf(hadoopConf, HiveConf.class); + } catch (Exception | NoClassDefFoundError e) { + LOG.warn("Fail to create Hive Configuration", e); + } + return hiveConf; + } + + @Override + public boolean delegationTokensRequired() throws Exception { + /** + * The general rule how a provider/receiver must behave is the following: The provider and + * the receiver must be added to the classpath together with all the additionally required + * dependencies. + * + * <p>This null check is required because the Hive provider is always on classpath but Hive + * jars are optional. Such case configuration is not able to be loaded. This construct is + * intended to be removed when Hive provider/receiver pair can be externalized (namely if a + * provider/receiver throws an exception then workload must be stopped). + */ + if (hiveConf == null) { + LOG.debug( + "Hive is not available (not packaged with this application), hence no " + + "tokens will be acquired."); + return false; + } + try { + if (!HadoopUtils.isKerberosSecurityEnabled(UserGroupInformation.getCurrentUser())) { + return false; + } + } catch (IOException e) { + LOG.debug("Hadoop Kerberos is not enabled."); + return false; + } + return !hiveConf.getTrimmed("hive.metastore.uris", "").isEmpty() + && kerberosLoginProvider.isLoginPossible(false); + } + + @Override + public ObtainedDelegationTokens obtainDelegationTokens() throws Exception { + UserGroupInformation freshUGI = kerberosLoginProvider.doLoginAndReturnUGI(); + return freshUGI.doAs( + (PrivilegedExceptionAction<ObtainedDelegationTokens>) + () -> { + Preconditions.checkNotNull(hiveConf); + Hive hive = Hive.get((HiveConf) hiveConf); + try { + LOG.info("Obtaining Kerberos security token for Hive"); + + String principal = + hiveConf.getTrimmed( + "hive.metastore.kerberos.principal", ""); + + String tokenStr = + hive.getDelegationToken( + UserGroupInformation.getCurrentUser().getUserName(), + principal); + Token<DelegationTokenIdentifier> hive2Token = new Token<>(); + hive2Token.decodeFromUrlString(tokenStr); + + Credentials credentials = new Credentials(); + credentials.addToken(TOKEN_ALIAS, hive2Token); + + return new ObtainedDelegationTokens( + HadoopDelegationTokenConverter.serialize(credentials), + Optional.empty()); Review Comment: When I played around with this, I haven't find a way to get back any expiration time related info from Hive, so I resorted to have my own configuration key. If you happen to have an installation at hand, then it might worth to try out the method you have suggested. It looks promising as it depends on Hadoop. Still I am a bit concerned as I have not seen the expiration time set on the token, even when it was expiring -- 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...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
