@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());
}