[ST-Engine][Starter] Add seatunnel own engine starter and e2e

pom.xml

@@ -197,6 +197,7 @@
         <commons-io.version>2.11.0</commons-io.version>
         <commons-collections4.version>4.4</commons-collections4.version>
         <maven-assembly-plugin.version>3.3.0</maven-assembly-plugin.version>
+        <commons-compress.version>1.20</commons-compress.version>
         <protostuff.version>1.8.0</protostuff.version>
         <spark.scope>provided</spark.scope>
         <flink.scope>provided</flink.scope>
@@ -629,6 +630,12 @@
                 <version>${jackson.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-compress</artifactId>
+                <version>${commons-compress.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>com.fasterxml.jackson.module</groupId>
                 <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>

seatunnel-apis/seatunnel-api-base/pom.xml

@@ -36,10 +36,6 @@
             <version>${project.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.seatunnel</groupId>
-            <artifactId>seatunnel-config-shade</artifactId>
-        </dependency>
     </dependencies>
 
 </project>

seatunnel-core/README.md

@@ -5,4 +5,7 @@ This module is the seatunnel job entrypoint. Seatunnel jobs are started by the b
 
 - seatunnel-core-flink: The flink job starter.
 - seatunnel-core-flink-sql: The flink sql job starter.
-- seatunnel-core-spark: The spark job starter.
+- seatunnel-core-spark: The spark job starter.
+- seatunnel-spark-starter: The spark job starter for connector-v2.
+- seatunnel-flink-starter: The flink job starter for connector-v2.
+- seatunnel-starter: The seatunnel engine job starter for connector-v2.

seatunnel-core/pom.xml

@@ -44,15 +44,15 @@
                 <module>seatunnel-core-starter</module>
                 <module>seatunnel-flink-starter</module>
                 <module>seatunnel-spark-starter</module>
-                <module>seatunnel-seatunnel-starter</module>
+                <module>seatunnel-starter</module>
             </modules>
         </profile>
         <profile>
             <id>engine-all</id>
             <modules>
                 <module>seatunnel-core-base</module>
                 <module>seatunnel-core-starter</module>
-                <module>seatunnel-seatunnel-starter</module>
+                <module>seatunnel-starter</module>
             </modules>
         </profile>
         <profile>
@@ -65,7 +65,7 @@
                 <module>seatunnel-core-starter</module>
                 <module>seatunnel-flink-starter</module>
                 <module>seatunnel-spark-starter</module>
-                <module>seatunnel-seatunnel-starter</module>
+                <module>seatunnel-starter</module>
             </modules>
         </profile>
     </profiles>

seatunnel-core/seatunnel-core-starter/pom.xml

@@ -40,19 +40,19 @@
 
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
-            <artifactId>seatunnel-api-flink</artifactId>
+            <artifactId>seatunnel-api</artifactId>
             <version>${project.version}</version>
         </dependency>
 
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
-            <artifactId>seatunnel-api-spark</artifactId>
+            <artifactId>seatunnel-plugin-discovery</artifactId>
             <version>${project.version}</version>
         </dependency>
+
         <dependency>
-            <groupId>org.apache.seatunnel</groupId>
-            <artifactId>seatunnel-plugin-discovery</artifactId>
-            <version>${project.version}</version>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-compress</artifactId>
         </dependency>
 
         <dependency>

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/config/EnvironmentFactory.java

@@ -19,8 +19,6 @@
 
 import org.apache.seatunnel.apis.base.env.RuntimeEnv;
 import org.apache.seatunnel.common.constants.JobMode;
-import org.apache.seatunnel.flink.FlinkEnvironment;
-import org.apache.seatunnel.spark.SparkEnvironment;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
@@ -31,39 +29,28 @@
  *
  * @param <ENVIRONMENT> environment type
  */
-public class EnvironmentFactory<ENVIRONMENT extends RuntimeEnv> {
+public abstract class EnvironmentFactory<ENVIRONMENT extends RuntimeEnv> {
 
     private static final String PLUGIN_NAME_KEY = "plugin_name";
 
     private final Config config;
-    private final EngineType engine;
 
-    public EnvironmentFactory(Config config, EngineType engine) {
+    public EnvironmentFactory(Config config) {
         this.config = config;
-        this.engine = engine;
     }
 
     // todo:put this method into submodule to avoid dependency on the engine
     public synchronized ENVIRONMENT getEnvironment() {
         Config envConfig = config.getConfig("env");
-        boolean enableHive = checkIsContainHive();
-        ENVIRONMENT env;
-        switch (engine) {
-            case SPARK:
-                env = (ENVIRONMENT) new SparkEnvironment().setEnableHive(enableHive);
-                break;
-            case FLINK:
-                env = (ENVIRONMENT) new FlinkEnvironment();
-                break;
-            default:
-                throw new IllegalArgumentException("Engine: " + engine + " is not supported");
-        }
+        ENVIRONMENT env = newEnvironment();
         env.setConfig(envConfig)
             .setJobMode(getJobMode(envConfig)).prepare();
         return env;
     }
 
-    private boolean checkIsContainHive() {
+    protected abstract ENVIRONMENT newEnvironment();
+
+    protected boolean checkIsContainHive() {
         List<? extends Config> sourceConfigList = config.getConfigList(PluginType.SOURCE.getType());
         for (Config c : sourceConfigList) {
             if (c.getString(PLUGIN_NAME_KEY).toLowerCase().contains("hive")) {

seatunnel-core/seatunnel-flink-starter/pom.xml

@@ -42,12 +42,6 @@
             <version>${project.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.seatunnel</groupId>
-            <artifactId>seatunnel-api</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
             <artifactId>seatunnel-translation-flink</artifactId>

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/FlinkStarter.java

@@ -20,7 +20,6 @@
 import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.core.starter.Starter;
 import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
-import org.apache.seatunnel.core.starter.flink.config.FlinkJobType;
 import org.apache.seatunnel.core.starter.flink.utils.CommandLineUtils;
 
 import java.util.List;
@@ -44,7 +43,7 @@ public class FlinkStarter implements Starter {
     private final String appJar;
 
     FlinkStarter(String[] args) {
-        this.flinkCommandArgs = CommandLineUtils.parseCommandArgs(args, FlinkJobType.JAR);
+        this.flinkCommandArgs = CommandLineUtils.parseCommandArgs(args);
         // set the deployment mode, used to get the job jar path.
         Common.setDeployMode(flinkCommandArgs.getDeployMode());
         Common.setStarter(true);

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/SeatunnelFlink.java

@@ -22,13 +22,12 @@
 import org.apache.seatunnel.core.starter.exception.CommandException;
 import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
 import org.apache.seatunnel.core.starter.flink.command.FlinkCommandBuilder;
-import org.apache.seatunnel.core.starter.flink.config.FlinkJobType;
 import org.apache.seatunnel.core.starter.flink.utils.CommandLineUtils;
 
 public class SeatunnelFlink {
 
     public static void main(String[] args) throws CommandException {
-        FlinkCommandArgs flinkCommandArgs = CommandLineUtils.parseCommandArgs(args, FlinkJobType.JAR);
+        FlinkCommandArgs flinkCommandArgs = CommandLineUtils.parseCommandArgs(args);
         Command<FlinkCommandArgs> flinkCommand = new FlinkCommandBuilder()
             .buildCommand(flinkCommandArgs);
         Seatunnel.run(flinkCommand);