[ZEPPELIN-5669] Check pyflink folder existence in yarn application mode

### What is this PR for?

Trivial PR to check the existence of the folder, and throw a meaningful error when it doesn't exist.

### What type of PR is it?
[ Improvement ]

### Todos
* [ ] - Task

### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-5669

### How should this be tested?
* CI pass

### Screenshots (if appropriate)

### Questions:
* Does the licenses files need update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No

Author: Jeff Zhang <[email protected]>

Closes apache#4320 from zjffdu/ZEPPELIN-5669 and squashes the following commits:

769d0aa [Jeff Zhang] [ZEPPELIN-5669] Check pyflink folder existence in yarn application mode

flink/flink1.12-shims/src/main/java/org/apache/zeppelin/flink/Flink112Shims.java

@@ -125,20 +125,24 @@ public String getPyFlinkPythonPath(Properties properties) throws IOException {
     if ("yarn-application".equalsIgnoreCase(mode)) {
       // for yarn application mode, FLINK_HOME is container working directory
       String flinkHome = new File(".").getAbsolutePath();
-      return getPyFlinkPythonPath(flinkHome + "/lib/python");
+      return getPyFlinkPythonPath(new File(flinkHome + "/lib/python"));
     }
 
     String flinkHome = System.getenv("FLINK_HOME");
     if (StringUtils.isNotBlank(flinkHome)) {
-      return getPyFlinkPythonPath(flinkHome + "/opt/python");
+      return getPyFlinkPythonPath(new File(flinkHome + "/opt/python"));
     } else {
       throw new IOException("No FLINK_HOME is specified");
     }
   }
 
-  private String getPyFlinkPythonPath(String pyFlinkFolder) {
+  private String getPyFlinkPythonPath(File pyFlinkFolder) throws IOException {
     LOGGER.info("Getting pyflink lib from {}", pyFlinkFolder);
-    List<File> depFiles = Arrays.asList(new File(pyFlinkFolder).listFiles());
+    if (!pyFlinkFolder.exists() || !pyFlinkFolder.isDirectory()) {
+      throw new IOException(String.format("PyFlink folder %s does not exist or is not a folder",
+              pyFlinkFolder.getAbsolutePath()));
+    }
+    List<File> depFiles = Arrays.asList(pyFlinkFolder.listFiles());
     StringBuilder builder = new StringBuilder();
     for (File file : depFiles) {
       LOGGER.info("Adding extracted file {} to PYTHONPATH", file.getAbsolutePath());

flink/flink1.13-shims/src/main/java/org/apache/zeppelin/flink/Flink113Shims.java

@@ -128,20 +128,24 @@ public String getPyFlinkPythonPath(Properties properties) throws IOException {
     if ("yarn-application".equalsIgnoreCase(mode)) {
       // for yarn application mode, FLINK_HOME is container working directory
       String flinkHome = new File(".").getAbsolutePath();
-      return getPyFlinkPythonPath(flinkHome + "/lib/python");
+      return getPyFlinkPythonPath(new File(flinkHome + "/lib/python"));
     }
 
     String flinkHome = System.getenv("FLINK_HOME");
     if (StringUtils.isNotBlank(flinkHome)) {
-      return getPyFlinkPythonPath(flinkHome + "/opt/python");
+      return getPyFlinkPythonPath(new File(flinkHome + "/opt/python"));
     } else {
       throw new IOException("No FLINK_HOME is specified");
     }
   }
 
-  private String getPyFlinkPythonPath(String pyFlinkFolder) {
+  private String getPyFlinkPythonPath(File pyFlinkFolder) throws IOException {
     LOGGER.info("Getting pyflink lib from {}", pyFlinkFolder);
-    List<File> depFiles = Arrays.asList(new File(pyFlinkFolder).listFiles());
+    if (!pyFlinkFolder.exists() || !pyFlinkFolder.isDirectory()) {
+      throw new IOException(String.format("PyFlink folder %s does not exist or is not a folder",
+              pyFlinkFolder.getAbsolutePath()));
+    }
+    List<File> depFiles = Arrays.asList(pyFlinkFolder.listFiles());
     StringBuilder builder = new StringBuilder();
     for (File file : depFiles) {
       LOGGER.info("Adding extracted file {} to PYTHONPATH", file.getAbsolutePath());

flink/flink1.14-shims/src/main/java/org/apache/zeppelin/flink/Flink114Shims.java

@@ -125,20 +125,24 @@ public String getPyFlinkPythonPath(Properties properties) throws IOException {
     if ("yarn-application".equalsIgnoreCase(mode)) {
       // for yarn application mode, FLINK_HOME is container working directory
       String flinkHome = new File(".").getAbsolutePath();
-      return getPyFlinkPythonPath(flinkHome + "/lib/python");
+      return getPyFlinkPythonPath(new File(flinkHome + "/lib/python"));
     }
 
     String flinkHome = System.getenv("FLINK_HOME");
     if (StringUtils.isNotBlank(flinkHome)) {
-      return getPyFlinkPythonPath(flinkHome + "/opt/python");
+      return getPyFlinkPythonPath(new File(flinkHome + "/opt/python"));
     } else {
       throw new IOException("No FLINK_HOME is specified");
     }
   }
 
-  private String getPyFlinkPythonPath(String pyFlinkFolder) {
+  private String getPyFlinkPythonPath(File pyFlinkFolder) throws IOException {
     LOGGER.info("Getting pyflink lib from {}", pyFlinkFolder);
-    List<File> depFiles = Arrays.asList(new File(pyFlinkFolder).listFiles());
+    if (!pyFlinkFolder.exists() || !pyFlinkFolder.isDirectory()) {
+      throw new IOException(String.format("PyFlink folder %s does not exist or is not a folder",
+              pyFlinkFolder.getAbsolutePath()));
+    }
+    List<File> depFiles = Arrays.asList(pyFlinkFolder.listFiles());
     StringBuilder builder = new StringBuilder();
     for (File file : depFiles) {
       LOGGER.info("Adding extracted file {} to PYTHONPATH", file.getAbsolutePath());