[Starter] [Core] Modify connector v2 starter

pom.xml

@@ -142,6 +142,7 @@
         <hadoop.binary.version>2.7</hadoop.binary.version>
         <jackson.version>2.12.6</jackson.version>
         <lombok.version>1.18.0</lombok.version>
+        <commons-compress.version>1.20</commons-compress.version>
         <mysql.version>8.0.16</mysql.version>
         <postgresql.version>42.3.3</postgresql.version>
         <dm-jdbc.version>8.1.2.141</dm-jdbc.version>
@@ -290,6 +291,13 @@
                     </exclusion>
                 </exclusions>
             </dependency>
+
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-compress</artifactId>
+                <version>${commons-compress.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>org.apache.logging.log4j</groupId>
                 <artifactId>log4j-core</artifactId>

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

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

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,27 @@
  *
  * @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,7 @@
 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.config.StarterConstant;
 import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
 
 import java.util.ArrayList;
@@ -46,7 +46,7 @@ public class FlinkStarter implements Starter {
     private final String appJar;
 
     FlinkStarter(String[] args) {
-        this.flinkCommandArgs = CommandLineUtils.parse(args, new FlinkCommandArgs(), FlinkJobType.JAR.getType(), true);
+        this.flinkCommandArgs = CommandLineUtils.parse(args, new FlinkCommandArgs(), StarterConstant.SHELL_NAME, true);
         // 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,13 @@
 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.config.StarterConstant;
 import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
 
 public class SeatunnelFlink {
 
     public static void main(String[] args) throws CommandException {
-        FlinkCommandArgs flinkCommandArgs = CommandLineUtils.parse(args, new FlinkCommandArgs(), FlinkJobType.JAR.getType(), true);
+        FlinkCommandArgs flinkCommandArgs = CommandLineUtils.parse(args, new FlinkCommandArgs(), StarterConstant.SHELL_NAME, true);
         Command<FlinkCommandArgs> flinkCommand = new FlinkCommandBuilder()
             .buildCommand(flinkCommandArgs);
         Seatunnel.run(flinkCommand);

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

@@ -0,0 +1,35 @@
+/*
+ * 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.core.starter.flink.config;
+
+import org.apache.seatunnel.core.starter.config.EnvironmentFactory;
+import org.apache.seatunnel.flink.FlinkEnvironment;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+public class FlinkEnvironmentFactory extends EnvironmentFactory<FlinkEnvironment> {
+
+    public FlinkEnvironmentFactory(Config config) {
+        super(config);
+    }
+
+    @Override
+    protected FlinkEnvironment newEnvironment() {
+        return new FlinkEnvironment();
+    }
+}

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/config/FlinkJobType.java → seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/config/StarterConstant.java

@@ -17,18 +17,8 @@
 
 package org.apache.seatunnel.core.starter.flink.config;
 
-public enum FlinkJobType {
-    JAR("start-seatunnel-flink.sh"),
-    SQL("start-seatunnel-sql.sh"),
-    ;
+public class StarterConstant {
 
-    private final String type;
+    public static final String SHELL_NAME = "start-seatunnel-flink-connector-v2.sh";
 
-    FlinkJobType(String type) {
-        this.type = type;
-    }
-
-    public String getType() {
-        return this.type;
-    }
 }