[flink] 02/05: [FLINK-12868][yarn] Fix yarn cluster can not be deployed if plugins dir does not exist

[pn...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit d17d4d8821c2e780a2c3aa9e194b97a2d18e797e
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Jun 17 09:37:56 2019 +0200

    [FLINK-12868][yarn] Fix yarn cluster can not be deployed if plugins dir does not exist
    
    Plugins dir should be optional and cluster deployment should not fail if it doesn't exist.
---
 .../flink/yarn/AbstractYarnClusterDescriptor.java  | 33 ++++++++++++++--------
 .../flink/yarn/YarnClusterDescriptorTest.java      | 22 +++++++++++++++
 2 files changed, 43 insertions(+), 12 deletions(-)

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index 8b51968..5b2bc2a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -1515,30 +1515,39 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
 	}
 
 	protected void addEnvironmentFoldersToShipFiles(Collection<File> effectiveShipFiles) {
+		addLibFoldersToShipFiles(effectiveShipFiles);
+		addPluginsFoldersToShipFiles(effectiveShipFiles);
+	}
+
+	private void addLibFoldersToShipFiles(Collection<File> effectiveShipFiles) {
 		// Add lib folder to the ship files if the environment variable is set.
 		// This is for convenience when running from the command-line.
 		// (for other files users explicitly set the ship files)
 		String libDir = System.getenv().get(ENV_FLINK_LIB_DIR);
 		if (libDir != null) {
-			addEnvFolderToShipFiles(effectiveShipFiles, libDir, ENV_FLINK_LIB_DIR);
+			File directoryFile = new File(libDir);
+			if (directoryFile.isDirectory()) {
+				effectiveShipFiles.add(directoryFile);
+			} else {
+				throw new YarnDeploymentException("The environment variable '" + ENV_FLINK_LIB_DIR +
+					"' is set to '" + libDir + "' but the directory doesn't exist.");
+			}
 		} else if (this.shipFiles.isEmpty()) {
 			LOG.warn("Environment variable '{}' not set and ship files have not been provided manually. " +
 				"Not shipping any library files.", ENV_FLINK_LIB_DIR);
 		}
+	}
 
+	private void addPluginsFoldersToShipFiles(Collection<File> effectiveShipFiles) {
 		String pluginsDir = System.getenv().get(ENV_FLINK_PLUGINS_DIR);
 		if (pluginsDir != null) {
-			addEnvFolderToShipFiles(effectiveShipFiles, pluginsDir, ENV_FLINK_PLUGINS_DIR);
-		}
-	}
-
-	private void addEnvFolderToShipFiles(Collection<File> effectiveShipFiles, String directory, String environmentVariableName) {
-		File directoryFile = new File(directory);
-		if (directoryFile.isDirectory()) {
-			effectiveShipFiles.add(directoryFile);
-		} else {
-			throw new YarnDeploymentException("The environment variable '" + environmentVariableName +
-				"' is set to '" + directory + "' but the directory doesn't exist.");
+			File directoryFile = new File(pluginsDir);
+			if (directoryFile.isDirectory()) {
+				effectiveShipFiles.add(directoryFile);
+			} else {
+				LOG.warn("The environment variable '" + ENV_FLINK_PLUGINS_DIR +
+					"' is set to '" + pluginsDir + "' but the directory doesn't exist.");
+			}
 		}
 	}
 
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index 4ac9839..842ebb6 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -45,6 +45,7 @@ import org.junit.rules.TemporaryFolder;
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -485,6 +486,27 @@ public class YarnClusterDescriptorTest extends TestLogger {
 		}
 	}
 
+	@Test
+	public void testEnvironmentEmptyPluginsShipping() throws Exception {
+		try (YarnClusterDescriptor descriptor = createYarnClusterDescriptor()) {
+			File pluginsFolder = Paths.get(temporaryFolder.getRoot().getAbsolutePath(), "s0m3_p4th_th4t_sh0uld_n0t_3x1sts").toFile();
+			Set<File> effectiveShipFiles = new HashSet<>();
+
+			final Map<String, String> oldEnv = System.getenv();
+			try {
+				Map<String, String> env = new HashMap<>(1);
+				env.put(ConfigConstants.ENV_FLINK_PLUGINS_DIR, pluginsFolder.getAbsolutePath());
+				CommonTestUtils.setEnv(env);
+				// only execute part of the deployment to test for shipped files
+				descriptor.addEnvironmentFoldersToShipFiles(effectiveShipFiles);
+			} finally {
+				CommonTestUtils.setEnv(oldEnv);
+			}
+
+			assertTrue(effectiveShipFiles.isEmpty());
+		}
+	}
+
 	/**
 	 * Tests that the YarnClient is only shut down if it is not shared.
 	 */