[FLINK-21700][security] Set service name as credential alias

Author: zuston

docs/layouts/shortcodes/generated/security_auth_kerberos_section.html

@@ -12,7 +12,7 @@
             <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 the delegation tokens for external services the Flink job needs to contact. Only HDFS and HBase are supported. It is used in Yarn deployments. If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers.  If false, Flink will assume that the delegation tokens are managed outside of Flink. As a consequence, it will not fetch delegation tokens for HDFS and HBase. You may need to disable this option, if you rely on submission mechanisms, e.g. Apache Oozie, to handle delegation tokens.</td>
+            <td>Indicates whether to fetch the delegation tokens for external services the Flink job needs to contact. Only HDFS and HBase are supported. It is used in Yarn deployments. If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers. If false, Flink will assume that the delegation tokens are managed outside of Flink. As a consequence, it will not fetch delegation tokens for HDFS and HBase. You may need to disable this option, if you rely on submission mechanisms, e.g. Apache Oozie, to handle delegation tokens.</td>
         </tr>
         <tr>
             <td><h5>security.kerberos.login.contexts</h5></td>

docs/layouts/shortcodes/generated/security_configuration.html

@@ -18,7 +18,7 @@
             <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 the delegation tokens for external services the Flink job needs to contact. Only HDFS and HBase are supported. It is used in Yarn deployments. If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers.  If false, Flink will assume that the delegation tokens are managed outside of Flink. As a consequence, it will not fetch delegation tokens for HDFS and HBase. You may need to disable this option, if you rely on submission mechanisms, e.g. Apache Oozie, to handle delegation tokens.</td>
+            <td>Indicates whether to fetch the delegation tokens for external services the Flink job needs to contact. Only HDFS and HBase are supported. It is used in Yarn deployments. If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers. If false, Flink will assume that the delegation tokens are managed outside of Flink. As a consequence, it will not fetch delegation tokens for HDFS and HBase. You may need to disable this option, if you rely on submission mechanisms, e.g. Apache Oozie, to handle delegation tokens.</td>
         </tr>
         <tr>
             <td><h5>security.kerberos.krb5-conf.path</h5></td>

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

@@ -116,7 +116,7 @@ public class SecurityOptions {
                     .withDescription(
                             "Indicates whether to fetch the delegation tokens for external services the Flink job needs to contact. "
                                     + "Only HDFS and HBase are supported. It is used in Yarn deployments. "
-                                    + "If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers.  "
+                                    + "If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers. "
                                     + "If false, Flink will assume that the delegation tokens are managed outside of Flink. "
                                     + "As a consequence, it will not fetch delegation tokens for HDFS and HBase. "
                                     + "You may need to disable this option, if you rely on submission mechanisms, e.g. Apache Oozie, "

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

@@ -34,7 +34,6 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.DataOutputBuffer;
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -219,9 +218,8 @@ public static void setTokensFor(
 
         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);
+            LOG.info("Adding user token " + token.getService() + " with " + token);
+            credentials.addToken(token.getService(), token);
         }
         try (DataOutputBuffer dob = new DataOutputBuffer()) {
             credentials.writeTokenStorageToStream(dob);
@@ -570,8 +568,7 @@ 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);
+                        taskManagerCred.addToken(token.getService(), token);
                     }
                 }