[FLINK-21700][yarn]Use delegation token's service name as alias

Author: zuston

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

@@ -219,9 +219,9 @@ 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);
+            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);
@@ -570,8 +570,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);
                     }
                 }