[FLINK-21700][yarn]Allow to disable fetching Hadoop delegation token on Yarn

Author: zuston

docs/layouts/shortcodes/generated/security_auth_kerberos_section.html

@@ -8,6 +8,12 @@
         </tr>
     </thead>
     <tbody>
+        <tr>
+            <td><h5>security.kerberos.fetch.delegation-token</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Indicates whether to fetch delegation token. If true, Flink will fetch HDFS/HBase delegation tokens and inject into AM container.If false, Flink will assume that the job has delegation tokens and will not fetch them.This applies to submission mechanisms like Oozie, which will obtain delegation tokens before submitting Flink Job.</td>
+        </tr>
         <tr>
             <td><h5>security.kerberos.login.contexts</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/security_configuration.html

@@ -14,6 +14,12 @@
             <td>List&lt;String&gt;</td>
             <td>List of factories that should be used to instantiate a security context. If multiple are configured, Flink will use the first compatible factory. You should have a NoOpSecurityContextFactory in this list as a fallback.</td>
         </tr>
+        <tr>
+            <td><h5>security.kerberos.fetch.delegation-token</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Indicates whether to fetch delegation token. If true, Flink will fetch HDFS/HBase delegation tokens and inject into AM container.If false, Flink will assume that the job has delegation tokens and will not fetch them.This applies to submission mechanisms like Oozie, which will obtain delegation tokens before submitting Flink Job.</td>
+        </tr>
         <tr>
             <td><h5>security.kerberos.krb5-conf.path</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java

@@ -108,6 +108,18 @@ public class SecurityOptions {
                                     + " (for example, `Client,KafkaClient` to use the credentials for ZooKeeper authentication and for"
                                     + " Kafka authentication)");
 
+    @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS)
+    public static final ConfigOption<Boolean> KERBEROS_FETCH_DELEGATION_TOKEN =
+            key("security.kerberos.fetch.delegation-token")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Indicates whether to fetch delegation token. If true, Flink will fetch "
+                                    + "HDFS/HBase delegation tokens and inject into AM container."
+                                    + "If false, Flink will assume that the job has delegation tokens and will not fetch them."
+                                    + "This applies to submission mechanisms like Oozie, which will obtain delegation tokens"
+                                    + " before submitting Flink Job.");
+
     // ------------------------------------------------------------------------
     //  ZooKeeper Security Options
     // ------------------------------------------------------------------------

flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java

@@ -197,13 +197,23 @@ private static LocalResource registerLocalResource(
     }
 
     public static void setTokensFor(
-            ContainerLaunchContext amContainer, List<Path> paths, Configuration conf)
+            ContainerLaunchContext amContainer,
+            List<Path> paths,
+            Configuration conf,
+            boolean obtainingDelegationTokens)
             throws IOException {
         Credentials credentials = new Credentials();
-        // for HDFS
-        TokenCache.obtainTokensForNamenodes(credentials, paths.toArray(new Path[0]), conf);
-        // for HBase
-        obtainTokenForHBase(credentials, conf);
+
+        if (obtainingDelegationTokens) {
+            LOG.info("Obtaining delegation tokens for HDFS and HBase.");
+            // for HDFS
+            TokenCache.obtainTokensForNamenodes(credentials, paths.toArray(new Path[0]), conf);
+            // for HBase
+            obtainTokenForHBase(credentials, conf);
+        } else {
+            LOG.info("Disable obtaining delegation tokens.");
+        }
+
         // for user
         UserGroupInformation currUsr = UserGroupInformation.getCurrentUser();
 

flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java

@@ -105,6 +105,7 @@
 import java.net.URI;
 import java.net.URLDecoder;
 import java.nio.charset.Charset;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -529,6 +530,17 @@ private ClusterClientProvider<ApplicationId> deployInternal(
                         "Hadoop security with Kerberos is enabled but the login user "
                                 + "does not have Kerberos credentials or delegation tokens!");
             }
+
+            boolean kerberosFetchDTEnabled =
+                    flinkConfiguration.getBoolean(SecurityOptions.KERBEROS_FETCH_DELEGATION_TOKEN);
+            boolean yarnAccessFSEnabled =
+                    flinkConfiguration.get(YarnConfigOptions.YARN_ACCESS) != null;
+            if (!kerberosFetchDTEnabled && yarnAccessFSEnabled) {
+                throw new RuntimeException(
+                        "Disable fetching kerberos delegation tokens but setting "
+                                + "yarn.security.kerberos.additionalFileSystems option!"
+                                + "These two options are mutually exclusive.");
+            }
         }
 
         isReadyForDeployment(clusterSpecification);
@@ -1079,15 +1091,24 @@ private ApplicationReport startAppMaster(
 
         // setup security tokens
         if (UserGroupInformation.isSecurityEnabled()) {
-            // set HDFS delegation tokens when security is enabled
-            LOG.info("Adding delegation token to the AM container.");
-            List<Path> yarnAccessList =
-                    ConfigUtils.decodeListFromConfig(
-                            configuration, YarnConfigOptions.YARN_ACCESS, Path::new);
+            List<Path> yarnAccessList = new ArrayList<>();
+
+            Boolean kerberosFetchDelegationTokenEnabled =
+                    configuration.getBoolean(SecurityOptions.KERBEROS_FETCH_DELEGATION_TOKEN);
+
+            if (kerberosFetchDelegationTokenEnabled) {
+                // set HDFS delegation tokens when security is enabled
+                LOG.info("Adding delegation token to the AM container.");
+                yarnAccessList =
+                        ConfigUtils.decodeListFromConfig(
+                                configuration, YarnConfigOptions.YARN_ACCESS, Path::new);
+            }
+
             Utils.setTokensFor(
                     amContainer,
                     ListUtils.union(yarnAccessList, fileUploader.getRemotePaths()),
-                    yarnConfiguration);
+                    yarnConfiguration,
+                    kerberosFetchDelegationTokenEnabled);
         }
 
         amContainer.setLocalResources(fileUploader.getRegisteredLocalResources());