fill up remote class path

Author: todd5167

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

@@ -288,18 +288,18 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 Table regTable = tableEnv.fromDataStream(adaptStream, fields);
                 tableEnv.registerTable(tableInfo.getName(), regTable);
                 registerTableCache.put(tableInfo.getName(), regTable);
-                classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SourceTableInfo.SOURCE_SUFFIX, remoteSqlPluginPath));
+                classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SourceTableInfo.SOURCE_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));
             } else if (tableInfo instanceof TargetTableInfo) {
 
                 TableSink tableSink = StreamSinkFactory.getTableSink((TargetTableInfo) tableInfo, localSqlPluginPath);
                 TypeInformation[] flinkTypes = FlinkUtil.transformTypes(tableInfo.getFieldClasses());
                 tableEnv.registerTableSink(tableInfo.getName(), tableInfo.getFields(), flinkTypes, tableSink);
-                classPathSet.add( PluginUtil.getRemoteJarFilePath(tableInfo.getType(), TargetTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
+                classPathSet.add( PluginUtil.getRemoteJarFilePath(tableInfo.getType(), TargetTableInfo.TARGET_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));
             } 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.getRemoteSideJarFilePath(tableInfo.getType(), sideOperator, SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
+                classPathSet.add(PluginUtil.getRemoteSideJarFilePath(tableInfo.getType(), sideOperator, SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));
             }else {
                 throw new RuntimeException("not support table type:" + tableInfo.getType());
             }

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

@@ -107,19 +107,19 @@ public static Properties stringToProperties(String str) throws IOException{
         return properties;
     }
 
-    public static URL getRemoteJarFilePath(String pluginType, String tableType, String remoteSqlRootDir) throws Exception {
+    public static URL getRemoteJarFilePath(String pluginType, String tableType, String remoteSqlRootDir, String localSqlPluginPath) throws Exception {
         String dirName = pluginType + tableType.toLowerCase();
         String prefix = String.format("%s-%s", pluginType, tableType.toLowerCase());
-        String jarPath = remoteSqlRootDir + SP + dirName;
-        String jarName = getCoreJarFileName(jarPath, prefix, false);
+        String jarPath = localSqlPluginPath + SP + dirName;
+        String jarName = getCoreJarFileName(jarPath, prefix);
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
-    public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir) throws Exception {
+    public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir, String localSqlPluginPath) throws Exception {
         String dirName = pluginType + sideOperator + tableType.toLowerCase();
         String prefix = String.format("%s-%s-%s", pluginType, sideOperator, tableType.toLowerCase());
-        String jarPath = remoteSqlRootDir + SP + dirName;
-        String jarName = getCoreJarFileName(jarPath, prefix, false);
+        String jarPath = localSqlPluginPath + SP + dirName;
+        String jarName = getCoreJarFileName(jarPath, prefix);
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
@@ -144,11 +144,7 @@ public static void addPluginJar(String pluginDir, DtClassLoader classLoader) thr
         }
     }
 
-    public static String getCoreJarFileName (String path, String prefix, boolean existCheck) throws Exception {
-        if (!existCheck) {
-            return prefix.toLowerCase() + ".jar";
-        }
-
+    public static String getCoreJarFileName (String path, String prefix) throws Exception {
         String coreJarFileName = null;
         File pluginDir = new File(path);
         if (pluginDir.exists() && pluginDir.isDirectory()){

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

@@ -59,7 +59,7 @@ public class LauncherMain {
 
 
     private static String getLocalCoreJarPath(String localSqlRootJar) throws Exception {
-        String jarPath = PluginUtil.getCoreJarFileName(localSqlRootJar, CORE_JAR, true);
+        String jarPath = PluginUtil.getCoreJarFileName(localSqlRootJar, CORE_JAR);
         String corePath = localSqlRootJar + SP + jarPath;
         return corePath;
     }