[FLINK-6376] [yarn] Improve inline code comments related to HDFS dele…

…gation token inclusion

This closes apache#3776.

flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/HadoopModule.java

@@ -60,18 +60,19 @@ public void install(SecurityUtils.SecurityConfiguration securityConfig) throws S
 				// supplement with any available tokens
 				String fileLocation = System.getenv(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION);
 				if (fileLocation != null) {
-					/*
-					 * Use reflection API since the API semantics are not available in Hadoop1 profile. Below APIs are
-					 * used in the context of reading the stored tokens from UGI.
-					 * Credentials cred = Credentials.readTokenStorageFile(new File(fileLocation), config.hadoopConf);
-					 * loginUser.addCredentials(cred);
-					 * Notify:If UGI use the keytab for login, do not load HDFS delegation token.
-					*/
+					// Use reflection API since the API semantics are not available in Hadoop1 profile. Below APIs are
+					// used in the context of reading the stored tokens from UGI.
+					// Credentials cred = Credentials.readTokenStorageFile(new File(fileLocation), config.hadoopConf);
+					// loginUser.addCredentials(cred);
 					try {
 						Method readTokenStorageFileMethod = Credentials.class.getMethod("readTokenStorageFile",
 							File.class, org.apache.hadoop.conf.Configuration.class);
 						Credentials cred = (Credentials) readTokenStorageFileMethod.invoke(null, new File(fileLocation),
 							securityConfig.getHadoopConfiguration());
+
+						// if UGI uses Kerberos keytabs for login, do not load HDFS delegation token since
+						// the UGI would prefer the delegation token instead, which eventually expires
+						// and does not fallback to using Kerberos tickets
 						Method getAllTokensMethod = Credentials.class.getMethod("getAllTokens");
 						Credentials credentials = new Credentials();
 						final Text HDFS_DELEGATION_TOKEN_KIND = new Text("HDFS_DELEGATION_TOKEN");
@@ -83,6 +84,7 @@ public void install(SecurityUtils.SecurityConfiguration securityConfig) throws S
 								credentials.addToken(id, token);
 							}
 						}
+
 						Method addCredentialsMethod = UserGroupInformation.class.getMethod("addCredentials",
 							Credentials.class);
 						addCredentialsMethod.invoke(loginUser, credentials);

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

@@ -794,7 +794,7 @@ public ApplicationReport startAppMaster(JobGraph jobGraph, YarnClient yarnClient
 		final ContainerLaunchContext amContainer = setupApplicationMasterContainer(hasLogback, hasLog4j, hasKrb5);
 
 		if (UserGroupInformation.isSecurityEnabled()) {
-			//set tokens when security is enable
+			// set HDFS delegation tokens when security is enabled
 			LOG.info("Adding delegation token to the AM container..");
 			Utils.setTokensFor(amContainer, paths, conf);
 		}

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

@@ -457,14 +457,13 @@ static ContainerLaunchContext createTaskExecutorContext(
 
 		try (DataOutputBuffer dob = new DataOutputBuffer()) {
 			log.debug("Adding security tokens to Task Executor Container launch Context....");
-			/*
-			 * For taskmanager yarn container context, read the tokens from the jobmanager yarn container local flie.
-			 * Notify: must read the tokens from the local file, but not from UGI context.Because if UGI is login
-			 * from Keytab, there is no HDFS degegation token in UGI context.
-			 */
+
+			// For TaskManager YARN container context, read the tokens from the jobmanager yarn container local flie.
+			// NOTE: must read the tokens from the local file, not from the UGI context, because if UGI is login
+			// using Kerberos keytabs, there is no HDFS delegation token in the UGI context.
 			String fileLocation = System.getenv(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION);
-			Method readTokenStorageFileMethod = Credentials.class.getMethod("readTokenStorageFile",
-				File.class, org.apache.hadoop.conf.Configuration.class);
+			Method readTokenStorageFileMethod = Credentials.class.getMethod(
+				"readTokenStorageFile", File.class, org.apache.hadoop.conf.Configuration.class);
 			Credentials cred = (Credentials) readTokenStorageFileMethod.invoke(null, new File(fileLocation),
 				new SecurityUtils.SecurityConfiguration(flinkConfig).getHadoopConfiguration());
 			cred.writeTokenStorageToStream(dob);