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

Change-Id: Id2468edf652ee5c2c4b76aab6b703c6bb18ec30c

Author: zuston

docs/layouts/shortcodes/generated/yarn_config_configuration.html

@@ -140,6 +140,12 @@
             <td>List&lt;String&gt;</td>
             <td>A comma-separated list of additional Kerberos-secured Hadoop filesystems Flink is going to access. For example, yarn.security.kerberos.additionalFileSystems=hdfs://namenode2:9002,hdfs://namenode3:9003. The client submitting to YARN needs to have access to these file systems to retrieve the security tokens.</td>
         </tr>
+        <tr>
+            <td><h5>yarn.security.kerberos.fetch.delegationToken.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>When this is true Flink will fetch HDFS/HBase delegation token injected into AM container.</td>
+        </tr>
         <tr>
             <td><h5>yarn.security.kerberos.localized-keytab-path</h5></td>
             <td style="word-wrap: break-word;">"krb5.keytab"</td>

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

@@ -197,21 +197,28 @@ private static LocalResource registerLocalResource(
     }
 
     public static void setTokensFor(
-            ContainerLaunchContext amContainer, List<Path> paths, Configuration conf)
+            ContainerLaunchContext amContainer,
+            List<Path> paths,
+            Configuration conf,
+            boolean yarnFetchDelegationEnabled)
             throws IOException {
         Credentials credentials = new Credentials();
-        // for HDFS
-        TokenCache.obtainTokensForNamenodes(credentials, paths.toArray(new Path[0]), conf);
-        // for HBase
-        obtainTokenForHBase(credentials, conf);
+
+        if (yarnFetchDelegationEnabled) {
+            // for HDFS
+            TokenCache.obtainTokensForNamenodes(credentials, paths.toArray(new Path[0]), conf);
+            // for HBase
+            obtainTokenForHBase(credentials, conf);
+        }
+
         // for user
         UserGroupInformation currUsr = UserGroupInformation.getCurrentUser();
 
         Collection<Token<? extends TokenIdentifier>> usrTok = currUsr.getTokens();
         for (Token<? extends TokenIdentifier> token : usrTok) {
-            final Text id = new Text(token.getIdentifier());
-            LOG.info("Adding user token " + id + " with " + token);
-            credentials.addToken(id, token);
+            final Text alias = new Text(token.getService());
+            LOG.info("Adding user token " + alias + " with " + token);
+            credentials.addToken(alias, token);
         }
         try (DataOutputBuffer dob = new DataOutputBuffer()) {
             credentials.writeTokenStorageToStream(dob);
@@ -560,8 +567,8 @@ static ContainerLaunchContext createTaskExecutorContext(
                 Collection<Token<? extends TokenIdentifier>> userTokens = cred.getAllTokens();
                 for (Token<? extends TokenIdentifier> token : userTokens) {
                     if (!token.getKind().equals(AMRMTokenIdentifier.KIND_NAME)) {
-                        final Text id = new Text(token.getIdentifier());
-                        taskManagerCred.addToken(id, token);
+                        final Text alias = new Text(token.getService());
+                        taskManagerCred.addToken(alias, token);
                     }
                 }
 

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;
@@ -1079,15 +1080,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 yarnFetchDelegationTokenEnabled =
+                    configuration.getBoolean(YarnConfigOptions.YARN_SECURITY_ENABLED);
+
+            if (yarnFetchDelegationTokenEnabled) {
+                // 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,
+                    yarnFetchDelegationTokenEnabled);
         }
 
         amContainer.setLocalResources(fileUploader.getRegisteredLocalResources());

flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java

@@ -344,6 +344,13 @@ public class YarnConfigOptions {
                     .withDescription(
                             "A comma-separated list of additional Kerberos-secured Hadoop filesystems Flink is going to access. For example, yarn.security.kerberos.additionalFileSystems=hdfs://namenode2:9002,hdfs://namenode3:9003. The client submitting to YARN needs to have access to these file systems to retrieve the security tokens.");
 
+    public static final ConfigOption<Boolean> YARN_SECURITY_ENABLED =
+            key("yarn.security.kerberos.fetch.delegationToken.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "When this is true Flink will fetch HDFS/HBase delegation token injected into AM container.");
+
     @SuppressWarnings("unused")
     public static final ConfigOption<String> HADOOP_CONFIG_KEY =
             key("flink.hadoop.<key>")