[FLINK-20267][runtime] The JaasModule didn't support symbolic links. …

…This is fixed now.

A test was added to verify the change.

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

@@ -23,6 +23,7 @@
 import org.apache.flink.runtime.security.DynamicConfiguration;
 import org.apache.flink.runtime.security.KerberosUtils;
 import org.apache.flink.runtime.security.SecurityConfiguration;
+import org.apache.flink.util.FileUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -159,12 +160,20 @@ private static File generateDefaultConfigFile(String workingDir) {
 		checkArgument(workingDir != null, "working directory should not be null.");
 		final File jaasConfFile;
 		try {
-			Path path = Files.createDirectories(Paths.get(workingDir));
+			Path path = Paths.get(workingDir);
+			if (Files.notExists(Paths.get(workingDir))) {
+				// We intentionally favored Path.toRealPath over Files.readSymbolicLinks as the latter one might return a
+				// relative path if the symbolic link refers to it. Path.toRealPath resolves the relative path instead.
+				Path parent = path.getParent().toRealPath();
+				Path resolvedPath = Paths.get(parent.toString(), path.getFileName().toString());
+
+				path = Files.createDirectories(resolvedPath);
+			}
 			Path jaasConfPath = Files.createTempFile(path, "jaas-", ".conf");
 			try (InputStream resourceStream = JaasModule.class.getClassLoader().getResourceAsStream(JAAS_CONF_RESOURCE_NAME)) {
 				Files.copy(resourceStream, jaasConfPath, StandardCopyOption.REPLACE_EXISTING);
 			}
-			jaasConfFile = jaasConfPath.toFile();
+			jaasConfFile = new File(workingDir, jaasConfPath.getFileName().toString());
 			jaasConfFile.deleteOnExit();
 		} catch (IOException e) {
 			throw new RuntimeException("unable to generate a JAAS configuration file", e);

flink-runtime/src/test/java/org/apache/flink/runtime/security/modules/JaasModuleTest.java

@@ -21,16 +21,23 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.runtime.security.SecurityConfiguration;
+import org.apache.flink.util.StringUtils;
 
+import org.hamcrest.core.StringStartsWith;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Random;
 
 import static org.apache.flink.runtime.security.modules.JaasModule.JAVA_SECURITY_AUTH_LOGIN_CONFIG;
+import static org.hamcrest.core.StringStartsWith.*;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -59,10 +66,21 @@ public void testJaasModuleFilePathIfWorkingDirNotPresent() throws IOException {
 		testJaasModuleFilePath(file.toPath().toString() + "/tmp");
 	}
 
+	@Test
+	public void testJaasModuleFilePathIfWorkingDirIsSymLink() throws IOException {
+		File baseFolder = folder.newFolder();
+		File actualFolder = new File(baseFolder, "actual_folder");
+		assertTrue(actualFolder.mkdir());
+
+		Path symlink = new File(baseFolder, "symlink").toPath();
+		Files.createSymbolicLink(symlink, actualFolder.toPath());
+		testJaasModuleFilePath(symlink.toString());
+	}
+
 	/**
 	 * Test that the jaas config file is created in the working directory.
 	 */
-	private void testJaasModuleFilePath(String workingDir) {
+	private void testJaasModuleFilePath(String workingDir) throws IOException {
 		Configuration configuration = new Configuration();
 		// set the string for CoreOptions.TMP_DIRS to mock the working directory.
 		configuration.setString(CoreOptions.TMP_DIRS, workingDir);
@@ -79,7 +97,7 @@ private void testJaasModuleFilePath(String workingDir) {
 	 * if we do not manually specify it.
 	 */
 	@Test
-	public void testCreateJaasModuleFileInTemporary() {
+	public void testCreateJaasModuleFileInTemporary() throws IOException {
 		Configuration configuration = new Configuration();
 		SecurityConfiguration sc = new SecurityConfiguration(configuration);
 		JaasModule module = new JaasModule(sc);
@@ -89,8 +107,12 @@ public void testCreateJaasModuleFileInTemporary() {
 		assertJaasFileLocateInRightDirectory(CoreOptions.TMP_DIRS.defaultValue());
 	}
 
-	private void assertJaasFileLocateInRightDirectory(String directory) {
-		assertTrue(System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG).startsWith(directory));
+	private void assertJaasFileLocateInRightDirectory(String directory) throws IOException {
+		String resolvedExpectedPath = new File(directory).toPath().toRealPath().toString();
+		String resolvedActualPathWithFile = new File(System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG)).toPath().toRealPath().toString();
+		assertThat("The resolved configured directory does not match the expected resolved one.", resolvedActualPathWithFile, startsWith(resolvedExpectedPath));
+
+		assertThat("The configured directory does not match the expected one.", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG), startsWith(directory));
 	}
 }