add classpath to cache file

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.parser.CreateFuncParser;
 import com.dtstack.flink.sql.parser.InsertSqlParser;
 import com.dtstack.flink.sql.side.SideSqlExec;
@@ -83,6 +84,8 @@
 
 public class Main {
 
+    private static final String CLASS_FILE_NAME_FMT = "class_path_%d";
+
     private static final ObjectMapper objMapper = new ObjectMapper();
 
     private static final Logger LOG = LoggerFactory.getLogger(Main.class);
@@ -272,15 +275,22 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 classPathSet.add( PluginUtil.getRemoteJarFilePath(tableInfo.getType(), TargetTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
             } else if(tableInfo instanceof SideTableInfo){
 
+                String sideOperator = ECacheType.ALL.name().equals(((SideTableInfo) tableInfo).getCacheType()) ? "all" : "async";
                 sideTableMap.put(tableInfo.getName(), (SideTableInfo) tableInfo);
-                classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
+                classPathSet.add(PluginUtil.getRemoteSideJarFilePath(tableInfo.getType(), sideOperator, SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
             }else {
                 throw new RuntimeException("not support table type:" + tableInfo.getType());
             }
         }
 
         //The plug-in information corresponding to the table is loaded into the classPath env
         addEnvClassPath(env, classPathSet);
+        int i = 0;
+        for(URL url : classPathSet){
+            String classFileName = String.format(CLASS_FILE_NAME_FMT, i);
+            env.registerCachedFile(url.getPath(),  classFileName, true);
+            i++;
+        }
     }
 
     private static StreamExecutionEnvironment getStreamExeEnv(Properties confProperties, String deployMode) throws IOException {

core/src/main/java/com/dtstack/flink/sql/util/PluginUtil.java

@@ -94,12 +94,18 @@ public static Properties stringToProperties(String str) throws IOException{
         return properties;
     }
 
-    public static URL getRemoteJarFilePath(String pluginType, String tableType,String remoteSqlRootDir) throws MalformedURLException {
+    public static URL getRemoteJarFilePath(String pluginType, String tableType, String remoteSqlRootDir) throws MalformedURLException {
         String dirName = pluginType + tableType.toLowerCase();
         String jarName = String.format("%s-%s.jar", pluginType, tableType.toLowerCase());
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
+    public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir) throws MalformedURLException {
+        String dirName = pluginType + sideOperator + tableType.toLowerCase();
+        String jarName = String.format("%s-%s-%s.jar", pluginType, sideOperator, tableType.toLowerCase());
+        return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
+    }
+
     public static String upperCaseFirstChar(String str){
         return str.substring(0, 1).toUpperCase() + str.substring(1);
     }

launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java

@@ -107,6 +107,7 @@ public boolean accept(File dir, String name) {
                     }
 
                     YarnClient yarnClient = YarnClient.createYarnClient();
+                    haYarnConf(yarnConf);
                     yarnClient.init(yarnConf);
                     yarnClient.start();
                     ApplicationId applicationId = null;
@@ -138,16 +139,12 @@ public boolean accept(File dir, String name) {
 
                     }
 
-                    if(org.apache.commons.lang3.StringUtils.isEmpty(applicationId.toString())) {
+                    if(StringUtils.isEmpty(applicationId.toString())) {
                         throw new RuntimeException("No flink session found on yarn cluster.");
                     }
 
-                    AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(config, yarnConf, ".", yarnClient, false);
-                    Field confField = AbstractYarnClusterDescriptor.class.getDeclaredField("conf");
-                    confField.setAccessible(true);
-                    haYarnConf(yarnConf);
-                    confField.set(clusterDescriptor, yarnConf);
 
+                    AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(config, yarnConf, ".", yarnClient, false);
                     ClusterClient clusterClient = clusterDescriptor.retrieve(applicationId);
                     clusterClient.setDetached(true);
                     return clusterClient;