YarnTaskManagerRunnerFactory { public static Runner create( String[] args, final Class<? extends YarnTaskManager> taskManager, Map<String, String> envs) throws IOException { EnvironmentInformation.logEnvironmentInfo(LOG, "YARN TaskManager", args); SignalHandler.register(LOG); JvmShutdownSafeguard.installAsShutdownHook(LOG); final Configuration configuration; try { configuration = TaskManager.parseArgsAndLoadConfig(args); } catch (Throwable t) { LOG.error(t.getMessage(), t); System.exit(TaskManager.STARTUP_FAILURE_RETURN_CODE()); return null; } final String yarnClientUsername = envs.get(YarnConfigKeys.ENV_HADOOP_USER_NAME); final String localDirs = envs.get(Environment.LOCAL_DIRS.key()); LOG.info("Current working/local Directory: {}", localDirs); final String currDir = envs.get(Environment.PWD.key()); LOG.info("Current working Directory: {}", currDir); final String remoteKeytabPrincipal = envs.get(YarnConfigKeys.KEYTAB_PRINCIPAL); LOG.info("TM: remoteKeytabPrincipal obtained {}", remoteKeytabPrincipal); 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); UserGroupInformation currentUser = UserGroupInformation.getCurrentUser(); LOG.info("YARN daemon is running as: {} Yarn client user obtainer: {}", currentUser.getShortUserName(), yarnClientUsername); String containerID = Preconditions.checkNotNull(envs.get(YarnFlinkResourceManager.ENV_FLINK_CONTAINER_ID)); final ResourceID resourceId = new ResourceID(containerID); LOG.info("ResourceID assigned for this container: {}", resourceId); File f = new File(currDir, Utils.KEYTAB_FILE_NAME); if (remoteKeytabPrincipal != null && f.exists()) { configuration.setString(SecurityOptions.KERBEROS_LOGIN_KEYTAB, f.getAbsolutePath()); configuration.setString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL, remoteKeytabPrincipal); } try { 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); } SecurityUtils.install(sc); return new Runner(configuration, resourceId, taskManager); } catch (Exception e) { LOG.error("Exception occurred while building Task Manager runner", e); throw new RuntimeException(e); } }  static Runner create(
			String[] args,
			final Class<? extends YarnTaskManager> taskManager,
			Map<String, String> envs);  }
@Test public void testKerberosKeytabConfiguration() throws IOException { final String resourceDirPath = Paths.get("src", "test", "resources").toAbsolutePath().toString(); final Map<String, String> envs = new HashMap<>(); envs.put(YarnFlinkResourceManager.ENV_FLINK_CONTAINER_ID, "test_container_00001"); envs.put(YarnConfigKeys.KEYTAB_PRINCIPAL, "testuser1@domain"); envs.put(ApplicationConstants.Environment.PWD.key(), resourceDirPath); final YarnTaskManagerRunnerFactory.Runner tmRunner = YarnTaskManagerRunnerFactory.create( new String[]{"--configDir", resourceDirPath}, YarnTaskManager.class, envs); final List<SecurityModule> modules = SecurityUtils.getInstalledModules(); Optional<SecurityModule> moduleOpt = modules.stream().filter(s -> s instanceof HadoopModule).findFirst(); if (moduleOpt.isPresent()) { HadoopModule hadoopModule = (HadoopModule) moduleOpt.get(); assertEquals("testuser1@domain", hadoopModule.getSecurityConfig().getPrincipal()); assertEquals(resourceDirPath + "/" + Utils.KEYTAB_FILE_NAME, hadoopModule.getSecurityConfig().getKeytab()); } else { fail("Can not find HadoopModule!"); } final Configuration configuration = tmRunner.getConfiguration(); assertEquals(resourceDirPath + "/" + Utils.KEYTAB_FILE_NAME, configuration.getString(SecurityOptions.KERBEROS_LOGIN_KEYTAB)); assertEquals("testuser1@domain", configuration.getString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL)); assertEquals("test_container_00001", tmRunner.getResourceId().toString()); }