Merge branch '1.10_release_4.1.x' into 1.10_release_4.2.x

Author: gituser

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -104,7 +104,7 @@ public class ExecuteProcessHelper {
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
     private static final String TIME_ZONE = "timezone";
-    private static final String PLUGIN_PATH_STR = "pluginPath";
+    private static final String PLUGIN_LOCAL_STR = "pluginPath";
     private static final String PLUGIN_LOAD_STR = "pluginLoadMode";
 
     public static FlinkPlanner flinkPlanner = new FlinkPlanner();
@@ -137,8 +137,8 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
             dirtyProperties.put(PLUGIN_LOAD_STR, pluginLoadMode);
         }
 
-        if (!pluginLoadMode.equalsIgnoreCase(EPluginLoadMode.LOCALTEST.name()) && Objects.isNull(dirtyProperties.get(PLUGIN_PATH_STR))) {
-            dirtyProperties.put(PLUGIN_PATH_STR, localSqlPluginPath);
+        if (!pluginLoadMode.equalsIgnoreCase(EPluginLoadMode.LOCALTEST.name()) && Objects.isNull(dirtyProperties.get(PLUGIN_LOCAL_STR))) {
+            dirtyProperties.put(PLUGIN_LOCAL_STR, localSqlPluginPath);
         }
 
         List<URL> jarUrlList = getExternalJarUrls(options.getAddjar());
@@ -393,9 +393,10 @@ public static Set<URL> registerTable(
             return Sets.newHashSet();
         }
         pluginClassPathSets.add(PluginUtil.buildDirtyPluginUrl(
-               String.valueOf(dirtyProperties.get(DirtyKeys.PLUGIN_TYPE_STR)),
-               String.valueOf(dirtyProperties.get(DirtyKeys.PLUGIN_PATH_STR)),
-               String.valueOf(dirtyProperties.get(DirtyKeys.PLUGIN_LOAD_MODE_STR))
+            String.valueOf(dirtyProperties.get(DirtyKeys.PLUGIN_TYPE_STR)),
+            localSqlPluginPath,
+            remoteSqlPluginPath,
+            pluginLoadMode
         ));
         return pluginClassPathSets;
     }

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

@@ -32,6 +32,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.net.MalformedURLException;
+import java.net.URI;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -49,12 +50,14 @@
 
 public class PluginUtil {
 
-    private static String SP = File.separator;
+    private static final String SP = File.separator;
 
     private static final String JAR_SUFFIX = ".jar";
 
     private static final String CLASS_PRE_STR = "com.dtstack.flink.sql";
 
+    private static final String DIRTY_DATA_PRE = "dirtyData";
+
     private static final Logger LOG = LoggerFactory.getLogger(PluginUtil.class);
 
     private static final ObjectMapper objectMapper = new ObjectMapper();
@@ -161,27 +164,32 @@ public static URL buildFinalJarFilePath(String pluginType, String tableType, Str
      * build dirty data url from plugin path
      *
      * @param dirtyType      type of dirty type
-     * @param pluginPath     plugin path
+     * @param localPath      local path
+     * @param remotePath     remote path
      * @param pluginLoadMode load plugin mode
      * @return dirty plugin url
      * @throws Exception exception
      */
     public static URL buildDirtyPluginUrl(
-            String dirtyType,
-            String pluginPath,
-            String pluginLoadMode) throws Exception {
+        String dirtyType,
+        String localPath,
+        String remotePath,
+        String pluginLoadMode) throws Exception {
         if (Objects.isNull(dirtyType)) {
             dirtyType = DirtyConsumerFactory.DEFAULT_DIRTY_TYPE;
         }
         String prefix = String.format("dirtyConsumer-%s", dirtyType.toLowerCase()).toLowerCase();
         String consumerType = DirtyConsumerFactory.DIRTY_CONSUMER_PATH + File.separator + dirtyType;
-        String consumerJar = PluginUtil.getJarFileDirPath(consumerType, pluginPath, pluginLoadMode);
+        String consumerJar = PluginUtil.getJarFileDirPath(consumerType, localPath, pluginLoadMode);
         String jarFileName = PluginUtil.getCoreJarFileName(
-                consumerJar,
-                prefix,
-                pluginLoadMode
+            consumerJar,
+            prefix,
+            pluginLoadMode
         );
-        return new URL("file:" + consumerJar + SP + jarFileName);
+        String path = remotePath == null ? localPath : remotePath;
+        String dirtyPath = path + SP + DIRTY_DATA_PRE + SP + dirtyType;
+        URI uri = URI.create("file:" + dirtyPath + SP + jarFileName);
+        return uri.toURL();
     }
 
     public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir, String localSqlPluginPath, String pluginLoadMode) throws Exception {