Merge branch '1.19.1-develop' into rel/1.19.1_arenadata1

arenadata · Sep 9, 2024 · 2b000a3 · 2b000a3
2 parents 18ece46 + c0e72f4
commit 2b000a3
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 2 deletions.
diff --git a/...c/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java b/...c/main/java/org/apache/flink/table/security/token/HiveServer2DelegationTokenProvider.java
@@ -18,9 +18,12 @@
 
 package org.apache.flink.table.security.token;
 
+import org.apache.commons.lang3.StringUtils;
+
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.SecurityOptions;
 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;
@@ -56,6 +59,8 @@ public class HiveServer2DelegationTokenProvider implements DelegationTokenProvid
 
     private Long tokenRenewalInterval;
 
+    private String hiveTokenRenewerPrincipal;
+
     @Override
     public String serviceName() {
         return "HiveServer2";
@@ -65,6 +70,8 @@ public String serviceName() {
     public void init(Configuration configuration) throws Exception {
         hiveConf = getHiveConfiguration(configuration);
         kerberosLoginProvider = new KerberosLoginProvider(configuration);
+        hiveTokenRenewerPrincipal = configuration
+                .get(SecurityOptions.DELEGATION_TOKENS_HIVE_RENEWER, "");
     }
 
     private org.apache.hadoop.conf.Configuration getHiveConfiguration(Configuration conf) {
@@ -136,9 +143,11 @@ public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
                             try {
                                 LOG.info("Obtaining Kerberos security token for HiveServer2");
 
-                                String principal =
+                                String principal = !StringUtils.isEmpty(hiveTokenRenewerPrincipal) ?
+                                        hiveTokenRenewerPrincipal :
                                         hiveConf.getTrimmed(
-                                                "hive.metastore.kerberos.principal", "");
+                                                "hive.metastore.kerberos.principal",
+                                                "");
 
                                 String tokenStr =
                                         hive.getDelegationToken(

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
@@ -215,6 +215,14 @@ public class SecurityOptions {
                                     + "possible to disable that behavior if it somehow conflicts "
                                     + "with the application being run.");
 
+    @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 5)
+    public static final ConfigOption<String> DELEGATION_TOKENS_HIVE_RENEWER =
+            key("security.delegation.tokens.hive.renewer")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Specifies the principal under which the delegation tokens "
+                            + "for Hive will be renewed.");
+
     /**
      * Returns a view over the given configuration via which options can be set/retrieved for the
      * given provider.