[FLINK-12491][docs][configuration] Fix incorrect javadoc for path sep…

…arators of CoreOptions.TMP_DIRS
apache · May 12, 2019 · 15fc612 · 15fc612
1 parent 76ae39a
commit 15fc612
Show file tree

Hide file tree

Showing 4 changed files with 109 additions and 4 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -219,7 +219,7 @@ public final class ConfigConstants {
 
 	/**
 	 * The config parameter defining the directories for temporary files, separated by
-	 * ",", "|", or the system's {@link java.io.File#pathSeparator}.
+	 * "," or the system's {@link java.io.File#pathSeparator}.
 	 *
 	 * @deprecated Use {@link CoreOptions#TMP_DIRS} instead
 	 */

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
@@ -18,14 +18,19 @@
 
 package org.apache.flink.configuration;
 
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.time.Time;
+import org.apache.flink.util.Preconditions;
 
 import javax.annotation.Nonnull;
 
 import java.io.File;
+import java.util.Arrays;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static org.apache.flink.configuration.MetricOptions.SYSTEM_RESOURCE_METRICS;
 import static org.apache.flink.configuration.MetricOptions.SYSTEM_RESOURCE_METRICS_PROBING_INTERVAL;
@@ -37,6 +42,11 @@ public class ConfigurationUtils {
 
 	private static final String[] EMPTY = new String[0];
 
+	@VisibleForTesting
+	static final String[] PATH_SEPARATORS = {",", File.pathSeparator};
+
+	private static final Pattern PATH_SEPARATORS_PATTERN = compilePathSeparatorsAsPattern(PATH_SEPARATORS);
+
 	/**
 	 * Get job manager's heap memory. This method will check the new key
 	 * {@link JobManagerOptions#JOB_MANAGER_HEAP_MEMORY} and
@@ -132,9 +142,27 @@ public static Configuration createConfiguration(Properties properties) {
 		return configuration;
 	}
 
+	private static Pattern compilePathSeparatorsAsPattern(String... separators) {
+		String regex = Arrays.stream(separators).map(Pattern::quote).collect(Collectors.joining("|"));
+		return Pattern.compile(regex);
+	}
+
+	private static String[] splitPaths(Pattern pattern, @Nonnull String separatedPaths) {
+		return separatedPaths.isEmpty() ? EMPTY : pattern.split(separatedPaths);
+	}
+
+	@Nonnull
+	@VisibleForTesting
+	static String[] splitPaths(@Nonnull String separatedPaths, @Nonnull String... pathSeparators) {
+		Preconditions.checkArgument(pathSeparators.length != 0, "Empty path separators");
+		Pattern pattern = compilePathSeparatorsAsPattern(pathSeparators);
+		return splitPaths(pattern, separatedPaths);
+	}
+
 	@Nonnull
-	private static String[] splitPaths(@Nonnull String separatedPaths) {
-		return separatedPaths.length() > 0 ? separatedPaths.split(",|" + File.pathSeparator) : EMPTY;
+	@VisibleForTesting
+	static String[] splitPaths(@Nonnull String separatedPaths) {
+		return splitPaths(PATH_SEPARATORS_PATTERN, separatedPaths);
 	}
 
 	// Make sure that we cannot instantiate this class

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
@@ -210,7 +210,7 @@ public static String[] getParentFirstLoaderPatterns(Configuration config) {
 
 	/**
 	 * The config parameter defining the directories for temporary files, separated by
-	 * ",", "|", or the system's {@link java.io.File#pathSeparator}.
+	 * "," or the system's {@link java.io.File#pathSeparator}.
 	 */
 	@Documentation.OverrideDefault("'LOCAL_DIRS' on Yarn. '_FLINK_TMP_DIR' on Mesos. System.getProperty(\"java.io.tmpdir\") in standalone.")
 	public static final ConfigOption<String> TMP_DIRS =

diff --git a/flink-core/src/test/java/org/apache/flink/configuration/ConfigurationUtilsTest.java b/flink-core/src/test/java/org/apache/flink/configuration/ConfigurationUtilsTest.java
@@ -22,17 +22,32 @@
 
 import org.junit.Test;
 
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
+import static org.hamcrest.Matchers.arrayWithSize;
+import static org.hamcrest.Matchers.emptyArray;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 
 /**
  * Tests for the {@link ConfigurationUtils}.
  */
 public class ConfigurationUtilsTest extends TestLogger {
 
+	private static final String[] TESTING_PATHS = {
+		"C:\\\\",
+		"current\\\\leaf",
+		"/root/dir/./../spaced dir/dotted.d/leaf",
+	};
+
 	@Test
 	public void testPropertiesToConfiguration() {
 		final Properties properties = new Properties();
@@ -51,4 +66,66 @@ public void testPropertiesToConfiguration() {
 		assertThat(configuration.toMap().size(), is(properties.size()));
 	}
 
+	@Test
+	public void testSplitPathsForEmptyPath() {
+		String[] paths = ConfigurationUtils.splitPaths("");
+		assertThat(paths, emptyArray());
+	}
+
+	@Test
+	public void testSplitPathsForSupportedSeparators() {
+		for (String separator : ConfigurationUtils.PATH_SEPARATORS) {
+			List<String> expectedPaths =  Arrays.stream(TESTING_PATHS)
+				.filter(p -> !p.contains(separator))
+				.filter(p -> !p.contains(File.pathSeparator))
+				.collect(Collectors.toList());
+			String separatedPaths = String.join(separator, expectedPaths);
+			String[] actualPaths = ConfigurationUtils.splitPaths(separatedPaths);
+			String message = String.format("Separator %s pattern separators %s",
+				separator, Arrays.toString(ConfigurationUtils.PATH_SEPARATORS));
+			assertEquals(message, expectedPaths, Arrays.asList(actualPaths));
+		}
+	}
+
+	@Test
+	public void testSplitPathsForNotSupportedSeparators() {
+		String[] notSupportedSeparators = Stream.of("|", ":", ";")
+			.filter(s -> !s.equals(File.pathSeparator))
+			.toArray(String[]::new);
+		for (String separator : notSupportedSeparators) {
+			List<String> expectedPaths =  Arrays.stream(TESTING_PATHS)
+				.filter(p -> !p.contains(separator))
+				.filter(p -> !p.contains(File.pathSeparator))
+				.collect(Collectors.toList());
+			String separatedPaths = String.join(separator, expectedPaths);
+			String[] actualPaths = ConfigurationUtils.splitPaths(separatedPaths);
+			String message = String.format("Pattern separators %s separatedPaths %s actualPaths: %s",
+				Arrays.toString(ConfigurationUtils.PATH_SEPARATORS), separatedPaths, Arrays.toString(actualPaths));
+			assertThat(message, actualPaths, arrayWithSize(1));
+			assertEquals(message, separatedPaths, actualPaths[0]);
+		}
+	}
+
+	@Test
+	public void testSplitPathsForCustomSeparators() {
+		String[] customSeparators = Stream.of(
+			Stream.of("|", "$"),
+			Stream.of(ConfigurationUtils.PATH_SEPARATORS),
+			// This way we can test both unix and windows path separator without ci test matrix.
+			Stream.of(":", ";")
+		)
+			.flatMap(Function.identity())
+			.distinct()
+			.toArray(String[]::new);
+		for (String separator : customSeparators) {
+			List<String> expectedPaths =  Arrays.stream(TESTING_PATHS)
+				.filter(p -> Arrays.stream(customSeparators).noneMatch(p::contains))
+				.collect(Collectors.toList());
+			String separatedPaths = String.join(separator, expectedPaths);
+			String[] actualPaths = ConfigurationUtils.splitPaths(separatedPaths, customSeparators);
+			String message = String.format("Separator %s pattern separators %s",
+				separator, Arrays.toString(customSeparators));
+			assertEquals(message, expectedPaths, Arrays.asList(actualPaths));
+		}
+	}
 }