class 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);

		// try to parse the command line arguments
		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;
		}

		// read the environment variables for YARN
		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);

		BootstrapTools.updateTmpDirectoriesInConfiguration(configuration, localDirs);

		// tell akka to die in case of an error
		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);

		// Infer the resource identifier from the environment variable
		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()) {
			// set keytab principal and replace path with the local path of the shipped keytab file in NodeManager
			configuration.setString(SecurityOptions.KERBEROS_LOGIN_KEYTAB, f.getAbsolutePath());
			configuration.setString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL, remoteKeytabPrincipal);
		}

		try {
			SecurityConfiguration 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);
		}

	}

}