[zeppelin] branch branch-0.9 updated: [ZEPPELIN-5021]. Add HIVE_CONF_DIR as resource in yarn interpreter mode for flink interpreter

[zjffdu]

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

zjffdu pushed a commit to branch branch-0.9
in repository https://gitbox.apache.org/repos/asf/zeppelin.git


The following commit(s) were added to refs/heads/branch-0.9 by this push:
     new f3d11c2  [ZEPPELIN-5021]. Add HIVE_CONF_DIR as resource in yarn 
interpreter mode for flink interpreter
f3d11c2 is described below

commit f3d11c226852687b2e4e401bc9b4dff206e4b337
Author: Jeff Zhang <zjf...@apache.org>
AuthorDate: Thu Aug 27 19:58:33 2020 +0800

    [ZEPPELIN-5021]. Add HIVE_CONF_DIR as resource in yarn interpreter mode for 
flink interpreter
    
    ### What is this PR for?
    
    This PR would add `HIVE_CONF_DIR` as yarn resources and set `HIVE_CONF_DIR` 
as `hive-conf` under working directory and AM container will download it and 
put it under the working directory.
    
    ### What type of PR is it?
    [Bug Fix ]
    
    ### Todos
    * [ ] - Task
    
    ### What is the Jira issue?
    * https://issues.apache.org/jira/browse/ZEPPELIN-5021
    
    ### How should this be tested?
    * Manually tested and 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 <zjf...@apache.org>
    
    Closes #3889 from zjffdu/ZEPPELIN-5021 and squashes the following commits:
    
    b54265104 [Jeff Zhang] [ZEPPELIN-5021]. Add HIVE_CONF_DIR as resource in 
yarn interpreter mode for flink interpreter
    
    (cherry picked from commit f6908f154d9698b587c8ba21ab8c60dce72f2a7b)
    Signed-off-by: Jeff Zhang <zjf...@apache.org>
---
 .../interpreter/launcher/YarnRemoteInterpreterProcess.java       | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git 
a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
 
b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
index 1f14f58..e095dd9 100644
--- 
a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
+++ 
b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
@@ -20,6 +20,7 @@ package org.apache.zeppelin.interpreter.launcher;
 import com.google.common.collect.Lists;
 import com.google.common.io.Files;
 import org.apache.commons.io.FileUtils;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -256,6 +257,13 @@ public class YarnRemoteInterpreterProcess extends 
RemoteInterpreterProcess {
       destPath = copyFileToRemote(stagingDir, srcPath, (short) 1);
       addResource(fs, destPath, localResources, LocalResourceType.ARCHIVE, 
"flink");
       FileUtils.forceDelete(flinkZip);
+
+      String hiveConfDir = 
launchContext.getProperties().getProperty("HIVE_CONF_DIR");
+      if (org.apache.commons.lang3.StringUtils.isBlank(hiveConfDir)) {
+        srcPath = localFs.makeQualified(new Path(new 
File(hiveConfDir).toURI()));
+        destPath = copyFileToRemote(stagingDir, srcPath, (short) 1);
+        addResource(fs, destPath, localResources, LocalResourceType.ARCHIVE, 
"hive_conf");
+      }
     }
     amContainer.setLocalResources(localResources);
 
@@ -296,6 +304,7 @@ public class YarnRemoteInterpreterProcess extends 
RemoteInterpreterProcess {
       this.envs.put("FLINK_CONF_DIR", ApplicationConstants.Environment.PWD.$() 
+ "/flink/conf");
       this.envs.put("FLINK_LIB_DIR", ApplicationConstants.Environment.PWD.$() 
+ "/flink/lib");
       this.envs.put("FLINK_PLUGINS_DIR", 
ApplicationConstants.Environment.PWD.$() + "/flink/plugins");
+      this.envs.put("HIVE_CONF_DIR", ApplicationConstants.Environment.PWD.$() 
+ "/hive_conf");
     }
     // set -Xmx
     int memory = Integer.parseInt(