[ST-Engine] Use SeatunnelChildFirstClassLoader to loader connector plugin

.gitignore

@@ -42,3 +42,5 @@ test.conf
 log4j.properties
 spark-warehouse
 *.flattened-pom.xml
+
+connectors

seatunnel-common/src/main/java/org/apache/seatunnel/common/config/Common.java

@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.common.config;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.io.File;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
@@ -97,14 +99,24 @@ public static Path connectorRootDir(String engine) {
      * Plugin Connector Jar Dir
      */
     public static Path connectorJarDir(String engine) {
-        return Paths.get(appRootDir().toString(), "connectors", engine.toLowerCase());
+        String seatunnelHome = System.getProperty("SEATUNNEL_HOME");
+        if (StringUtils.isBlank(seatunnelHome)) {
+            return Paths.get(appRootDir().toString(), "connectors", engine.toLowerCase());
+        } else {
+            return Paths.get(seatunnelHome, "connectors", engine.toLowerCase());
+        }
     }
 
     /**
      * Plugin Connector Dir
      */
     public static Path connectorDir() {
-        return Paths.get(appRootDir().toString(), "connectors");
+        String seatunnelHome = System.getProperty("SEATUNNEL_HOME");
+        if (StringUtils.isBlank(seatunnelHome)) {
+            return Paths.get(appRootDir().toString(), "connectors");
+        } else {
+            return Paths.get(seatunnelHome, "connectors");
+        }
     }
 
     public static Path pluginTarball() {

seatunnel-e2e/pom.xml

@@ -38,6 +38,7 @@
                 <module>seatunnel-flink-connector-v2-e2e</module>
                 <module>seatunnel-spark-connector-v2-e2e</module>
                 <module>seatunnel-flink-sql-e2e</module>
+                <module>seatunnel-engine-e2e</module>
             </modules>
             <dependencies>
                 <dependency>

seatunnel-e2e/seatunnel-engine-e2e/pom.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>seatunnel-e2e</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>seatunnel-engine-e2e</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.hazelcast</groupId>
+            <artifactId>hazelcast</artifactId>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-engine-client</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.seatunnel</groupId>
+                    <artifactId>connector-file-local</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.seatunnel</groupId>
+                    <artifactId>connector-fake</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-connectors-v2-dist</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-engine-server</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-e2e/seatunnel-engine-e2e/src/test/java/org/apache/seatunnel/engine/e2e/TestUtils.java

@@ -0,0 +1,83 @@
+/*
+ * 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.apache.seatunnel.engine.e2e;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.Arrays;
+
+public class TestUtils {
+    public static String getResource(String confFile) {
+        return System.getProperty("user.dir") + "/src/test/resources" + confFile;
+    }
+
+    public static void initPluginDir() {
+        // copy connectors to project_root/connectors dir
+        System.setProperty("SEATUNNEL_HOME", System.getProperty("user.dir") +
+            String.format("%s..%s..%s", File.separator, File.separator, File.separator));
+        File seatunnelRootDir = new File(System.getProperty("SEATUNNEL_HOME"));
+
+        File connectorDir = new File(seatunnelRootDir +
+            File.separator +
+            "connectors/seatunnel");
+
+        if (connectorDir.exists()) {
+            connectorDir.delete();
+        }
+
+        connectorDir.mkdirs();
+
+        File connectorDistDir = new File(
+            seatunnelRootDir +
+                File.separator +
+                "seatunnel-connectors-v2-dist" +
+                File.separator +
+                "target" +
+                File.separator +
+                "lib");
+
+        Arrays.stream(connectorDistDir.listFiles()).forEach(file -> {
+            if (file.getName().startsWith("connector-")) {
+                Path copied = Paths.get(connectorDir + File.separator + file.getName());
+                Path originalPath = file.toPath();
+                try {
+                    Files.copy(originalPath, copied, StandardCopyOption.REPLACE_EXISTING);
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+
+        Path targetPluginMappingFile = Paths.get(seatunnelRootDir +
+            File.separator +
+            "connectors" +
+            File.separator +
+            "plugin-mapping.properties");
+
+        Path sourcePluginMappingFile = Paths.get(seatunnelRootDir + File.separator + "plugin-mapping.properties");
+        try {
+            Files.copy(sourcePluginMappingFile, targetPluginMappingFile, StandardCopyOption.REPLACE_EXISTING);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

seatunnel-e2e/seatunnel-engine-e2e/src/test/java/org/apache/seatunnel/engine/e2e/engine/JobExecutionIT.java

@@ -0,0 +1,87 @@
+/*
+ * 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.apache.seatunnel.engine.e2e.engine;
+
+import org.apache.seatunnel.common.config.Common;
+import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
+import org.apache.seatunnel.engine.client.job.JobProxy;
+import org.apache.seatunnel.engine.common.config.ConfigProvider;
+import org.apache.seatunnel.engine.common.config.JobConfig;
+import org.apache.seatunnel.engine.common.config.SeaTunnelClientConfig;
+import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
+import org.apache.seatunnel.engine.core.job.JobStatus;
+import org.apache.seatunnel.engine.e2e.TestUtils;
+import org.apache.seatunnel.engine.server.SeaTunnelNodeContext;
+
+import com.google.common.collect.Lists;
+import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.instance.impl.HazelcastInstanceFactory;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutionException;
+
+public class JobExecutionIT {
+    private static final Logger LOGGER = LoggerFactory.getLogger(JobExecutionIT.class);
+
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+        SeaTunnelConfig seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig();
+        HazelcastInstanceFactory.newHazelcastInstance(seaTunnelConfig.getHazelcastConfig(),
+            Thread.currentThread().getName(),
+            new SeaTunnelNodeContext(ConfigProvider.locateAndGetSeaTunnelConfig()));
+    }
+
+    @Test
+    public void testSayHello() {
+        SeaTunnelClientConfig seaTunnelClientConfig = new SeaTunnelClientConfig();
+        seaTunnelClientConfig.getNetworkConfig().setAddresses(Lists.newArrayList("localhost:5801"));
+        SeaTunnelClient engineClient = new SeaTunnelClient(seaTunnelClientConfig);
+
+        String msg = "Hello world";
+        String s = engineClient.printMessageToMaster(msg);
+        Assert.assertEquals(msg, s);
+    }
+
+    @Test
+    public void testExecuteJob() {
+        TestUtils.initPluginDir();
+        Common.setDeployMode(DeployMode.CLIENT);
+        String filePath = TestUtils.getResource("/fakesource_to_file_complex.conf");
+        JobConfig jobConfig = new JobConfig();
+        jobConfig.setName("fake_to_file");
+
+        ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
+        SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig);
+        JobExecutionEnvironment jobExecutionEnv = engineClient.createExecutionContext(filePath, jobConfig);
+
+        JobProxy jobProxy = null;
+        try {
+            jobProxy = jobExecutionEnv.execute();
+            JobStatus jobStatus = jobProxy.waitForJobComplete();
+            Assert.assertEquals(JobStatus.FINISHED, jobStatus);
+        } catch (ExecutionException | InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

seatunnel-e2e/seatunnel-engine-e2e/src/test/resources/fakesource_to_file.conf

@@ -0,0 +1,65 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  # You can set flink configuration here
+  execution.parallelism = 1
+  job.mode = "STREAMING"
+  execution.checkpoint.interval = 5000
+  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+    FakeSource {
+      result_table_name = "fake"
+      field_name = "name,age",
+      parallelism = 3
+    }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+  LocalFile {
+    path="file:///tmp/hive/warehouse/test2"
+    field_delimiter="\t"
+    row_delimiter="\n"
+    partition_by=["age"]
+    partition_dir_expression="${k0}=${v0}"
+    is_partition_field_write_in_file=true
+    file_name_expression="${transactionId}_${now}"
+    file_format="text"
+    sink_columns=["name","age"]
+    filename_time_format="yyyy.MM.dd"
+    is_enable_transaction=true
+    save_mode="error"
+
+  }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+}