[Feature] [Flink] support_add_jar_other_url

dinky-admin/src/main/java/org/dinky/init/SystemInit.java

@@ -44,10 +44,12 @@
 import org.dinky.service.TenantService;
 import org.dinky.service.resource.impl.HdfsResourceManager;
 import org.dinky.service.resource.impl.OssResourceManager;
+import org.dinky.url.RsURLStreamHandlerFactory;
 import org.dinky.utils.JSONUtil;
 import org.dinky.utils.OssTemplate;
 import org.dinky.utils.UDFUtils;
 
+import org.apache.catalina.webresources.TomcatURLStreamHandlerFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 
@@ -104,6 +106,11 @@ public void run(ApplicationArguments args) {
         initDolphinScheduler();
         registerUDF();
         updateGitBuildState();
+        registerURL();
+    }
+
+    private void registerURL() {
+        TomcatURLStreamHandlerFactory.getInstance().addUserFactory(new RsURLStreamHandlerFactory());
     }
 
     private void initResources() {
@@ -162,7 +169,9 @@ private void initResources() {
                 }));
     }
 
-    /** init task monitor */
+    /**
+     * init task monitor
+     */
     private void initTaskMonitor() {
         List<JobInstance> jobInstances = jobInstanceService.listJobInstanceActive();
         List<DaemonTaskConfig> configList = new ArrayList<>();
@@ -173,7 +182,9 @@ private void initTaskMonitor() {
         DaemonFactory.start(configList);
     }
 
-    /** init DolphinScheduler */
+    /**
+     * init DolphinScheduler
+     */
     private void initDolphinScheduler() {
         systemConfiguration
                 .getAllConfiguration()

dinky-admin/src/main/java/org/dinky/service/resource/BaseResourceManager.java

@@ -23,6 +23,8 @@
 import org.dinky.service.resource.impl.HdfsResourceManager;
 import org.dinky.service.resource.impl.OssResourceManager;
 
+import java.io.InputStream;
+
 import org.springframework.web.multipart.MultipartFile;
 
 import cn.hutool.core.io.FileUtil;
@@ -39,6 +41,8 @@ public interface BaseResourceManager {
 
     String getFileContent(String path);
 
+    InputStream getFile(String path);
+
     static BaseResourceManager getInstance() {
         switch (SystemConfiguration.getInstances().getResourcesModel().getValue()) {
             case HDFS:
@@ -50,7 +54,7 @@ static BaseResourceManager getInstance() {
         }
     }
 
-    default String getFile(String path) {
+    default String getFilePath(String path) {
         return FileUtil.normalize(
                 FileUtil.file(instances.getResourcesUploadBasePath().getValue(), path)
                         .toString());

dinky-admin/src/main/java/org/dinky/service/resource/impl/HdfsResourceManager.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.fs.Path;
 
 import java.io.IOException;
+import java.io.InputStream;
 
 import org.springframework.web.multipart.MultipartFile;
 
@@ -38,7 +39,7 @@ public class HdfsResourceManager implements BaseResourceManager {
     @Override
     public void remove(String path) {
         try {
-            getHdfs().delete(new Path(getFile(path)), true);
+            getHdfs().delete(new Path(getFilePath(path)), true);
         } catch (IOException e) {
             throw BusException.valueOf("file.delete.failed", e);
         }
@@ -47,7 +48,7 @@ public void remove(String path) {
     @Override
     public void rename(String path, String newPath) {
         try {
-            getHdfs().rename(new Path(getFile(path)), new Path(getFile(newPath)));
+            getHdfs().rename(new Path(getFilePath(path)), new Path(getFilePath(newPath)));
         } catch (IOException e) {
             throw BusException.valueOf("file.rename.failed", e);
         }
@@ -56,7 +57,7 @@ public void rename(String path, String newPath) {
     @Override
     public void putFile(String path, MultipartFile file) {
         try {
-            FSDataOutputStream stream = getHdfs().create(new Path(getFile(path)), true);
+            FSDataOutputStream stream = getHdfs().create(new Path(getFilePath(path)), true);
             stream.write(file.getBytes());
             stream.flush();
             stream.close();
@@ -67,8 +68,13 @@ public void putFile(String path, MultipartFile file) {
 
     @Override
     public String getFileContent(String path) {
+        return IoUtil.readUtf8(getFile(path));
+    }
+
+    @Override
+    public InputStream getFile(String path) {
         try {
-            return IoUtil.readUtf8(getHdfs().open(new Path(getFile(path))));
+            return getHdfs().open(new Path(getFilePath(path)));
         } catch (IOException e) {
             throw BusException.valueOf("file.read.failed", e);
         }

dinky-admin/src/main/java/org/dinky/service/resource/impl/OssResourceManager.java

@@ -24,6 +24,8 @@
 import org.dinky.service.resource.BaseResourceManager;
 import org.dinky.utils.OssTemplate;
 
+import java.io.InputStream;
+
 import org.springframework.web.multipart.MultipartFile;
 
 import com.amazonaws.services.s3.model.CopyObjectRequest;
@@ -36,36 +38,41 @@ public class OssResourceManager implements BaseResourceManager {
 
     @Override
     public void remove(String path) {
-        getOssTemplate().removeObject(getOssTemplate().getBucketName(), getFile(path));
+        getOssTemplate().removeObject(getOssTemplate().getBucketName(), getFilePath(path));
     }
 
     @Override
     public void rename(String path, String newPath) {
         CopyObjectRequest copyObjectRequest = new CopyObjectRequest(
                 getOssTemplate().getBucketName(),
-                getFile(path),
+                getFilePath(path),
                 getOssTemplate().getBucketName(),
-                getFile(newPath));
+                getFilePath(newPath));
         getOssTemplate().getAmazonS3().copyObject(copyObjectRequest);
         DeleteObjectRequest deleteObjectRequest =
-                new DeleteObjectRequest(getOssTemplate().getBucketName(), getFile(path));
+                new DeleteObjectRequest(getOssTemplate().getBucketName(), getFilePath(path));
         getOssTemplate().getAmazonS3().deleteObject(deleteObjectRequest);
     }
 
     @Override
     public void putFile(String path, MultipartFile file) {
         try {
-            getOssTemplate().putObject(getOssTemplate().getBucketName(), getFile(path), file.getInputStream());
+            getOssTemplate().putObject(getOssTemplate().getBucketName(), getFilePath(path), file.getInputStream());
         } catch (Exception e) {
             throw new DinkyException(e);
         }
     }
 
     @Override
     public String getFileContent(String path) {
-        return IoUtil.readUtf8(getOssTemplate()
-                .getObject(getOssTemplate().getBucketName(), getFile(path))
-                .getObjectContent());
+        return IoUtil.readUtf8(getFile(path));
+    }
+
+    @Override
+    public InputStream getFile(String path) {
+        return getOssTemplate()
+                .getObject(getOssTemplate().getBucketName(), getFilePath(path))
+                .getObjectContent();
     }
 
     public OssTemplate getOssTemplate() {

dinky-admin/src/main/java/org/dinky/url/RsURLConnection.java

@@ -0,0 +1,60 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.url;
+
+import org.dinky.data.exception.BusException;
+import org.dinky.function.constant.PathConstant;
+import org.dinky.service.resource.BaseResourceManager;
+
+import java.io.File;
+import java.io.InputStream;
+import java.net.URL;
+import java.net.URLConnection;
+
+import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.StrUtil;
+
+public class RsURLConnection extends URLConnection {
+    private InputStream inputStream;
+
+    @Override
+    public void connect() {
+        BaseResourceManager instance = BaseResourceManager.getInstance();
+        if (instance == null) {
+            throw BusException.valueOf("ResourceManager is disabled");
+        }
+    }
+
+    @Override
+    public InputStream getInputStream() {
+        connect();
+        return inputStream;
+    }
+
+    public RsURLConnection(URL url) {
+        super(url);
+    }
+
+    public File toFile() {
+        connect();
+        String path = StrUtil.join(File.separator, PathConstant.TMP_PATH, "rs", getURL().getPath());
+        return FileUtil.writeFromStream(inputStream, path);
+    }
+}

dinky-admin/src/main/java/org/dinky/url/RsURLStreamHandler.java

@@ -0,0 +1,33 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.url;
+
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.net.URLStreamHandler;
+
+public class RsURLStreamHandler extends URLStreamHandler {
+
+    @Override
+    protected URLConnection openConnection(URL u) throws IOException {
+        return new RsURLConnection(u);
+    }
+}

dinky-admin/src/main/java/org/dinky/url/RsURLStreamHandlerFactory.java

@@ -0,0 +1,42 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.url;
+
+import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
+
+import java.net.URLStreamHandler;
+import java.net.URLStreamHandlerFactory;
+
+import cn.hutool.core.lang.Singleton;
+
+public class RsURLStreamHandlerFactory implements URLStreamHandlerFactory {
+    @Override
+    public URLStreamHandler createURLStreamHandler(String protocol) {
+        if ("rs".equals(protocol)) {
+            return new RsURLStreamHandler();
+        }
+        try {
+            Class.forName("org.apache.hadoop.fs.FsUrlStreamHandlerFactory");
+        } catch (Exception e) {
+            return null;
+        }
+        return Singleton.get(FsUrlStreamHandlerFactory.class).createURLStreamHandler(protocol);
+    }
+}

dinky-common/src/main/java/org/dinky/utils/URLUtils.java

@@ -22,12 +22,40 @@
 import java.io.File;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.net.URLConnection;
 import java.util.Arrays;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.StrUtil;
+
 /** @since 0.7.0 */
 public class URLUtils {
+    private static final String TMP_PATH = StrUtil.join(File.separator, System.getProperty("user.dir"), "tmp");
+    /**
+     * url download file to local
+     * @param urlPath urlPath
+     * @return file
+     */
+    public static File toFile(String urlPath) {
+        try {
+            URL url = new URL(urlPath);
+            URLConnection urlConnection = url.openConnection();
+            if ("http".equals(url.getProtocol())
+                    || "https".equals(url.getProtocol())
+                    || "hdfs".equals(url.getProtocol())
+                    || "rs".equals(url.getProtocol())) {
+                String path = StrUtil.join(File.separator, TMP_PATH, "downloadFile", url.getPath());
+                return FileUtil.writeFromStream(urlConnection.getInputStream(), path);
+            } else if ("file".equals(url.getProtocol())) {
+                return new File(url.getPath());
+            }
+            return null;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
 
     /**
      * 获得URL，常用于使用绝对路径时的情况

dinky-executor/src/main/java/org/dinky/parser/check/AddJarSqlParser.java

@@ -20,6 +20,7 @@
 package org.dinky.parser.check;
 
 import org.dinky.process.exception.DinkyException;
+import org.dinky.utils.URLUtils;
 
 import java.io.File;
 import java.util.HashSet;
@@ -29,7 +30,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.util.ReUtil;
 import cn.hutool.core.util.StrUtil;
 
@@ -51,11 +51,12 @@ public static Set<File> getAllFilePath(String[] statements) {
         patternStatements(statements).stream()
                 .map(x -> ReUtil.findAll(ADD_JAR_PATTERN, x, 2).get(0))
                 .distinct()
-                .forEach(path -> {
-                    if (!FileUtil.exist(path)) {
-                        throw new DinkyException(StrUtil.format("file : {} not exists!", path));
+                .forEach(urlPath -> {
+                    File file = URLUtils.toFile(urlPath);
+                    if (file == null || !file.exists()) {
+                        throw new DinkyException(StrUtil.format("file : {} not exists!", urlPath));
                     }
-                    fileSet.add(FileUtil.file(path));
+                    fileSet.add(file);
                 });
         return fileSet;
     }