flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
/**
	 * The instance entry point for the YARN task executor. Obtains user group information and calls
	 * the main work method {@link TaskManagerRunner#runTaskManager(Configuration, ResourceID)}  as a
	 * privileged action.
	 *
	 * @param args The command line arguments.
	 */
The instance entry point for the YARN task executor. Obtains user group information and calls
the main work method  TaskManagerRunner#runTaskManager(Configuration, ResourceID)  as a
privileged action.

@param args The command line arguments.
The instance entry point for the YARN task executor . Obtains user group information and calls the main work method TaskManagerRunner#runTaskManager ( Configuration ResourceID ) as a privileged action . @param args The command line arguments .
private static void run(String[] args) {
		try {
			LOG.debug("All environment variables: {}", ENV);
			final String yarnClientUsername = ENV.get(YarnConfigKeys.ENV_HADOOP_USER_NAME);
			final String localDirs = ENV.get(Environment.LOCAL_DIRS.key());
			LOG.info("Current working/local Directory: {}", localDirs);
			final String currDir = ENV.get(Environment.PWD.key());
			LOG.info("Current working Directory: {}", currDir);
			final String remoteKeytabPath = ENV.get(YarnConfigKeys.KEYTAB_PATH);
			LOG.info("TM: remote keytab path obtained {}", remoteKeytabPath);
			final String remoteKeytabPrincipal = ENV.get(YarnConfigKeys.KEYTAB_PRINCIPAL);
			LOG.info("TM: remote keytab principal obtained {}", remoteKeytabPrincipal);
			final Configuration configuration = GlobalConfiguration.loadConfiguration(currDir);
			FileSystem.initialize(configuration);
			String flinkTempDirs = configuration.getString(ConfigConstants.TASK_MANAGER_TMP_DIR_KEY, null);
			if (flinkTempDirs == null) {
				LOG.info("Setting directories for temporary file " + localDirs);
				configuration.setString(ConfigConstants.TASK_MANAGER_TMP_DIR_KEY, localDirs);
			}
			else {
				LOG.info("Overriding YARN's temporary file directories with those " +
						"specified in the Flink config: " + flinkTempDirs);
			}
			configuration.setBoolean(AkkaOptions.JVM_EXIT_ON_FATAL_ERROR, true);
			String keytabPath = null;
			if (remoteKeytabPath != null) {
				File f = new File(currDir, Utils.KEYTAB_FILE_NAME);
				keytabPath = f.getAbsolutePath();
				LOG.info("keytab path: {}", keytabPath);
			}
			UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
			LOG.info("YARN daemon is running as: {} Yarn client user obtainer: {}",
					currentUser.getShortUserName(), yarnClientUsername);
			SecurityConfiguration sc;
			File krb5Conf = new File(currDir, Utils.KRB5_FILE_NAME);
			if (krb5Conf.exists() && krb5Conf.canRead()) {
				String krb5Path = krb5Conf.getAbsolutePath();
				LOG.info("KRB5 Conf: {}", krb5Path);
				org.apache.hadoop.conf.Configuration hadoopConfiguration = new org.apache.hadoop.conf.Configuration();
				hadoopConfiguration.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, "kerberos");
				hadoopConfiguration.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, "true");
				sc = new SecurityConfiguration(configuration,
					Collections.singletonList(securityConfig -> new HadoopModule(securityConfig, hadoopConfiguration)));
			} else {
				sc = new SecurityConfiguration(configuration);
			}
			if (keytabPath != null && remoteKeytabPrincipal != null) {
				configuration.setString(SecurityOptions.KERBEROS_LOGIN_KEYTAB, keytabPath);
				configuration.setString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL, remoteKeytabPrincipal);
			}
			final String containerId = ENV.get(YarnFlinkResourceManager.ENV_FLINK_CONTAINER_ID);
			Preconditions.checkArgument(containerId != null,
				"ContainerId variable %s not set", YarnFlinkResourceManager.ENV_FLINK_CONTAINER_ID);
			final String taskExecutorHostname = ENV.get(YarnResourceManager.ENV_FLINK_NODE_ID);
			if (taskExecutorHostname != null) {
				configuration.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, taskExecutorHostname);
			}
			SecurityUtils.install(sc);
			SecurityUtils.getInstalledContext().runSecured(new Callable<Void>() {
				@Override
				public Void call() throws Exception {
					TaskManagerRunner.runTaskManager(configuration, new ResourceID(containerId));
					return null;
				}
			});
		}
		catch (Throwable t) {
			LOG.error("YARN TaskManager initialization failed.", t);
			System.exit(INIT_ERROR_EXIT_CODE);
		}
	}
private static void run ( String [ ] args ) { try { LOG . debug ( "All environment variables: {}" , ENV ) ; final String yarnClientUsername = ENV . get ( YarnConfigKeys . ENV_HADOOP_USER_NAME ) ; final String localDirs = ENV . get ( Environment . LOCAL_DIRS . key ( ) ) ; LOG . info ( "Current working/local Directory: {}" , localDirs ) ; final String currDir = ENV . get ( Environment . PWD . key ( ) ) ; LOG . info ( "Current working Directory: {}" , currDir ) ; final String remoteKeytabPath = ENV . get ( YarnConfigKeys . KEYTAB_PATH ) ; LOG . info ( "TM: remote keytab path obtained {}" , remoteKeytabPath ) ; final String remoteKeytabPrincipal = ENV . get ( YarnConfigKeys . KEYTAB_PRINCIPAL ) ; LOG . info ( "TM: remote keytab principal obtained {}" , remoteKeytabPrincipal ) ; final Configuration configuration = GlobalConfiguration . loadConfiguration ( currDir ) ; FileSystem . initialize ( configuration ) ; String flinkTempDirs = configuration . getString ( ConfigConstants . TASK_MANAGER_TMP_DIR_KEY , null ) ; if ( flinkTempDirs == null ) { LOG . info ( "Setting directories for temporary file " + localDirs ) ; configuration . setString ( ConfigConstants . TASK_MANAGER_TMP_DIR_KEY , localDirs ) ; } else { LOG . info ( "Overriding YARN's temporary file directories with those " + "specified in the Flink config: " + flinkTempDirs ) ; } configuration . setBoolean ( AkkaOptions . JVM_EXIT_ON_FATAL_ERROR , true ) ; String keytabPath = null ; if ( remoteKeytabPath != null ) { File f = new File ( currDir , Utils . KEYTAB_FILE_NAME ) ; keytabPath = f . getAbsolutePath ( ) ; LOG . info ( "keytab path: {}" , keytabPath ) ; } UserGroupInformation currentUser = UserGroupInformation . getCurrentUser ( ) ; LOG . info ( "YARN daemon is running as: {} Yarn client user obtainer: {}" , currentUser . getShortUserName ( ) , yarnClientUsername ) ; SecurityConfiguration sc ; File krb5Conf = new File ( currDir , Utils . KRB5_FILE_NAME ) ; if ( krb5Conf . exists ( ) && krb5Conf . canRead ( ) ) { String krb5Path = krb5Conf . getAbsolutePath ( ) ; LOG . info ( "KRB5 Conf: {}" , krb5Path ) ; org . apache . hadoop . conf . Configuration hadoopConfiguration = new org . apache . hadoop . conf . Configuration ( ) ; hadoopConfiguration . set ( CommonConfigurationKeysPublic . HADOOP_SECURITY_AUTHENTICATION , "kerberos" ) ; hadoopConfiguration . set ( CommonConfigurationKeysPublic . HADOOP_SECURITY_AUTHORIZATION , "true" ) ; sc = new SecurityConfiguration ( configuration , Collections . singletonList ( securityConfig -> new HadoopModule ( securityConfig , hadoopConfiguration ) ) ) ; } else { sc = new SecurityConfiguration ( configuration ) ; } if ( keytabPath != null && remoteKeytabPrincipal != null ) { configuration . setString ( SecurityOptions . KERBEROS_LOGIN_KEYTAB , keytabPath ) ; configuration . setString ( SecurityOptions . KERBEROS_LOGIN_PRINCIPAL , remoteKeytabPrincipal ) ; } final String containerId = ENV . get ( YarnFlinkResourceManager . ENV_FLINK_CONTAINER_ID ) ; Preconditions . checkArgument ( containerId != null , "ContainerId variable %s not set" , YarnFlinkResourceManager . ENV_FLINK_CONTAINER_ID ) ; final String taskExecutorHostname = ENV . get ( YarnResourceManager . ENV_FLINK_NODE_ID ) ; if ( taskExecutorHostname != null ) { configuration . setString ( ConfigConstants . TASK_MANAGER_HOSTNAME_KEY , taskExecutorHostname ) ; } SecurityUtils . install ( sc ) ; SecurityUtils . getInstalledContext ( ) . runSecured ( new Callable < Void > ( ) { @ Override public Void call ( ) throws Exception { TaskManagerRunner . runTaskManager ( configuration , new ResourceID ( containerId ) ) ; return null ; } } ) ; } catch ( Throwable t ) { LOG . error ( "YARN TaskManager initialization failed." , t ) ; System . exit ( INIT_ERROR_EXIT_CODE ) ; } }
The instance entry point for the YARN task executor.
The instance entry point for the YARN task executor .
// configure local directory // tell akka to die in case of an error //To support Yarn Secure Integration Test Scenario // use the hostname passed by job manager // make sure that everything whatever ends up in the log