[Feature][Flink] Add flink version 1.20.1 support

.github/workflows/backend.yml

@@ -722,7 +722,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 150
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -874,7 +874,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 150
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -1163,7 +1163,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -1238,7 +1238,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 140
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -1314,7 +1314,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -1339,7 +1339,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 150
+    timeout-minutes: 180
     steps:
       - name: Checkout repository
         uses: actions/checkout@v2
@@ -1419,7 +1419,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 180
+    timeout-minutes: 210
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -1444,7 +1444,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}

pom.xml

@@ -76,6 +76,7 @@
         <commons-logging.version>1.2</commons-logging.version>
         <flink.1.13.6.version>1.13.6</flink.1.13.6.version>
         <flink.1.15.3.version>1.15.3</flink.1.15.3.version>
+        <flink.1.20.1.version>1.20.1</flink.1.20.1.version>
         <spark.2.4.0.version>2.4.0</spark.2.4.0.version>
         <spark.3.3.0.version>3.3.0</spark.3.3.0.version>
         <spark.binary.2.4.version>2.4</spark.binary.2.4.version>

seatunnel-common/src/main/java/org/apache/seatunnel/common/constants/EngineType.java

@@ -23,6 +23,7 @@ public enum EngineType {
     SPARK3("spark", "seatunnel-spark-3-starter.jar", "start-seatunnel-spark-3-connector-v2.sh"),
     FLINK13("flink", "seatunnel-flink-13-starter.jar", "start-seatunnel-flink-13-connector-v2.sh"),
     FLINK15("flink", "seatunnel-flink-15-starter.jar", "start-seatunnel-flink-15-connector-v2.sh"),
+    FLINK20("flink", "seatunnel-flink-20-starter.jar", "start-seatunnel-flink-20-connector-v2.sh"),
     SEATUNNEL("seatunnel", "seatunnel-starter.jar", "seatunnel.sh");
 
     private final String engine;

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

@@ -35,6 +35,7 @@
     <modules>
         <module>seatunnel-flink-13-starter</module>
         <module>seatunnel-flink-15-starter</module>
+        <module>seatunnel-flink-20-starter</module>
         <module>seatunnel-flink-starter-common</module>
     </modules>
 

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

@@ -17,97 +17,20 @@
 
 package org.apache.seatunnel.core.starter.flink;
 
-import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.common.constants.EngineType;
-import org.apache.seatunnel.core.starter.Starter;
-import org.apache.seatunnel.core.starter.enums.MasterType;
-import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
-import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-
-/** The SeaTunnel flink starter, used to generate the final flink job execute command. */
-public class FlinkStarter implements Starter {
-    private static final String APP_NAME = SeaTunnelFlink.class.getName();
+/**
+ * The SeaTunnel flink starter for Flink 1.13, used to generate the final flink job execute command.
+ */
+public class FlinkStarter extends AbstractFlinkStarter {
     public static final String APP_JAR_NAME = EngineType.FLINK13.getStarterJarName();
-    public static final String SHELL_NAME = EngineType.FLINK13.getStarterShellName();
-    public static final String RUNTIME_FILE = "runtime.tar.gz";
-    private final FlinkCommandArgs flinkCommandArgs;
-    private final String appJar;
 
     FlinkStarter(String[] args) {
-        this.flinkCommandArgs =
-                CommandLineUtils.parse(args, new FlinkCommandArgs(), SHELL_NAME, true);
-        // set the deployment mode, used to get the job jar path.
-        Common.setDeployMode(flinkCommandArgs.getDeployMode());
-        Common.setStarter(true);
-        this.appJar = Common.appStarterDir().resolve(APP_JAR_NAME).toString();
+        super(args, EngineType.FLINK13);
     }
 
     public static void main(String[] args) {
         FlinkStarter flinkStarter = new FlinkStarter(args);
         System.out.println(String.join(" ", flinkStarter.buildCommands()));
     }
-
-    @Override
-    public List<String> buildCommands() {
-        List<String> command = new ArrayList<>();
-        // set start command
-        command.add("${FLINK_HOME}/bin/flink");
-        // set deploy mode, run or run-application
-        command.add(flinkCommandArgs.getDeployMode().getDeployMode());
-        // set submitted target master
-        if (flinkCommandArgs.getMasterType() != null) {
-            command.add("--target");
-            command.add(flinkCommandArgs.getMasterType().getMaster());
-        }
-        // set yarn application mode parameters
-        if (flinkCommandArgs.getMasterType() == MasterType.YARN_APPLICATION) {
-            command.add(
-                    String.format("-Dyarn.ship-files=\"%s\"", flinkCommandArgs.getConfigFile()));
-            command.add(String.format("-Dyarn.ship-archives=%s", RUNTIME_FILE));
-        }
-        // set yarn application name
-        if (flinkCommandArgs.getMasterType() == MasterType.YARN_APPLICATION
-                || flinkCommandArgs.getMasterType() == MasterType.YARN_PER_JOB
-                || flinkCommandArgs.getMasterType() == MasterType.YARN_SESSION) {
-            command.add(String.format("-Dyarn.application.name=%s", flinkCommandArgs.getJobName()));
-        }
-        // set flink original parameters
-        command.addAll(flinkCommandArgs.getOriginalParameters());
-        // set main class name
-        command.add("-c");
-        command.add(APP_NAME);
-        // set main jar name
-        command.add(appJar);
-        // set config file path
-        command.add("--config");
-        command.add(flinkCommandArgs.getConfigFile());
-        // set check config flag
-        if (flinkCommandArgs.isCheckConfig()) {
-            command.add("--check");
-        }
-        // set job name
-        command.add("--name");
-        command.add(flinkCommandArgs.getJobName());
-        // set encryption
-        if (flinkCommandArgs.isEncrypt()) {
-            command.add("--encrypt");
-        }
-        // set decryption
-        if (flinkCommandArgs.isDecrypt()) {
-            command.add("--decrypt");
-        }
-        // set deploy mode
-        command.add("--deploy-mode");
-        command.add(flinkCommandArgs.getDeployMode().getDeployMode());
-        // set extra system properties
-        flinkCommandArgs.getVariables().stream()
-                .filter(Objects::nonNull)
-                .map(String::trim)
-                .forEach(variable -> command.add("-i " + variable));
-        return command;
-    }
 }

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

@@ -18,19 +18,11 @@
 package org.apache.seatunnel.core.starter.flink;
 
 import org.apache.seatunnel.common.constants.EngineType;
-import org.apache.seatunnel.core.starter.SeaTunnel;
 import org.apache.seatunnel.core.starter.exception.CommandException;
-import org.apache.seatunnel.core.starter.flink.args.FlinkCommandArgs;
-import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
 
-public class SeaTunnelFlink {
+/** SeaTunnel Flink 1.13 main entry point. */
+public class SeaTunnelFlink extends AbstractSeaTunnelFlink {
     public static void main(String[] args) throws CommandException {
-        FlinkCommandArgs flinkCommandArgs =
-                CommandLineUtils.parse(
-                        args,
-                        new FlinkCommandArgs(),
-                        EngineType.FLINK13.getStarterShellName(),
-                        true);
-        SeaTunnel.run(flinkCommandArgs.buildCommand());
+        runSeaTunnel(args, EngineType.FLINK13);
     }
 }

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java

@@ -22,9 +22,6 @@
 import org.apache.seatunnel.common.constants.JobMode;
 import org.apache.seatunnel.core.starter.execution.RuntimeEnvironment;
 
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
 public class FlinkRuntimeEnvironment extends AbstractFlinkRuntimeEnvironment
         implements RuntimeEnvironment {
 

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java

@@ -46,7 +46,8 @@
 
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 
-import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.net.URL;
 import java.util.HashMap;
@@ -61,10 +62,11 @@
 import static org.apache.seatunnel.api.table.factory.FactoryUtil.discoverOptionalFactory;
 
 @SuppressWarnings({"unchecked", "rawtypes"})
-@Slf4j
 public class SinkExecuteProcessor
         extends FlinkAbstractPluginExecuteProcessor<Optional<? extends Factory>> {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(SinkExecuteProcessor.class);
+
     protected SinkExecuteProcessor(
             List<URL> jarPaths,
             Config envConfig,
@@ -167,7 +169,7 @@ public SeaTunnelSink tryGenerateMultiTableSink(
             ReadonlyConfig sinkConfig,
             ClassLoader classLoader) {
         if (sinks.values().stream().anyMatch(sink -> !(sink instanceof SupportMultiTableSink))) {
-            log.info("Unsupported multi table sink api, rollback to sink template");
+            LOGGER.info("Unsupported multi table sink api, rollback to sink template");
             // choose the first sink
             return sinks.values().iterator().next();
         }