[FLINK-7630] Allow passing a File or an InputStream to ParameterTool.fromPropertiesFile()

docs/dev/best_practices.md

@@ -47,8 +47,14 @@ The `ParameterTool` provides a set of predefined static methods for reading the
 
 The following method will read a [Properties](https://docs.oracle.com/javase/tutorial/essential/environment/properties.html) file and provide the key/value pairs:
 {% highlight java %}
-String propertiesFile = "/home/sam/flink/myjob.properties";
+String propertiesFilePath = "/home/sam/flink/myjob.properties";
+ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFilePath);
+
+File propertiesFile = new File(propertiesFilePath);
 ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFile);
+
+InputStream propertiesFileInputStream = new FileInputStream(file);
+ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFileInputStream);
 {% endhighlight %}
 
 

flink-java/src/main/java/org/apache/flink/api/java/utils/ParameterTool.java

@@ -33,6 +33,7 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Serializable;
 import java.util.Arrays;
@@ -155,13 +156,37 @@ public class ParameterTool extends ExecutionConfig.GlobalJobParameters implement
 	 */
 	public static ParameterTool fromPropertiesFile(String path) throws IOException {
 		File propertiesFile = new File(path);
-		if (!propertiesFile.exists()) {
-			throw new FileNotFoundException("Properties file " + propertiesFile.getAbsolutePath() + " does not exist");
+		return fromPropertiesFile(propertiesFile);
+	}
+
+	/**
+	 * Returns {@link ParameterTool} for the given {@link Properties} file.
+	 *
+	 * @param file File object to the properties file
+	 * @return A {@link ParameterTool}
+	 * @throws IOException If the file does not exist
+	 * @see Properties
+	 */
+	public static ParameterTool fromPropertiesFile(File file) throws IOException {
+		if (!file.exists()) {
+			throw new FileNotFoundException("Properties file " + file.getAbsolutePath() + " does not exist");
 		}
-		Properties props = new Properties();
-		try (FileInputStream fis = new FileInputStream(propertiesFile)) {
-			props.load(fis);
+		try (FileInputStream fis = new FileInputStream(file)) {
+			return fromPropertiesFile(fis);
 		}
+	}
+
+	/**
+	 * Returns {@link ParameterTool} for the given InputStream from {@link Properties} file.
+	 *
+	 * @param inputStream InputStream from the properties file
+	 * @return A {@link ParameterTool}
+	 * @throws IOException If the file does not exist
+	 * @see Properties
+	 */
+	public static ParameterTool fromPropertiesFile(InputStream inputStream) throws IOException {
+		Properties props = new Properties();
+		props.load(inputStream);
 		return fromMap((Map) props);
 	}
 

flink-java/src/test/java/org/apache/flink/api/java/utils/ParameterToolTest.java

@@ -24,6 +24,7 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
 import java.io.File;
+import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -121,6 +122,16 @@ public class ParameterToolTest extends AbstractParameterToolTest {
 		ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFile.getAbsolutePath());
 		Assert.assertEquals(2, parameter.getNumberOfParameters());
 		validate(parameter);
+
+		parameter = ParameterTool.fromPropertiesFile(propertiesFile);
+		Assert.assertEquals(2, parameter.getNumberOfParameters());
+		validate(parameter);
+
+		try (FileInputStream fis = new FileInputStream(propertiesFile)) {
+			parameter = ParameterTool.fromPropertiesFile(fis);
+		}
+		Assert.assertEquals(2, parameter.getNumberOfParameters());
+		validate(parameter);
 	}
 
 	@Test