[Feature-18691][task-flink] Submit Flink SQL through SQL Gateway

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/pom.xml

@@ -45,6 +45,14 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
+
+        <!-- Provided: the official bundle jar is deployed once under DS libs/, not shaded into this plugin. -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-jdbc-driver-bundle</artifactId>
+            <version>2.2.0</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
     <build>

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkParameters.java

@@ -17,11 +17,16 @@
 
 package org.apache.dolphinscheduler.plugin.task.flink;
 
+import org.apache.dolphinscheduler.plugin.task.api.enums.SqlSourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 
 import lombok.Data;
 
@@ -99,8 +104,67 @@ public class FlinkParameters extends AbstractParameters {
      */
     private String rawScript;
 
+    /**
+     * Absent or CLIENT keeps sql-client.sh. SQL_GATEWAY submits over JDBC.
+     */
+    private FlinkSqlSubmitType sqlSubmitType;
+
+    /**
+     * example: jdbc:flink://host:port
+     */
+    private String flinkJdbcUrl;
+
+    /**
+     * optional: username/password or other jdbc properties
+     */
+    private Map<String, String> jdbcProperties;
+
+    /**
+     * init script source, used only when {@link #sqlSubmitType} is SQL_GATEWAY
+     */
+    private SqlSourceType initScriptType = SqlSourceType.FILE;
+
+    /**
+     * resource list for init script file (when initScriptType=FILE), size 1
+     */
+    private List<ResourceInfo> initScriptResourceList;
+
+    /**
+     * main script source, used only when {@link #sqlSubmitType} is SQL_GATEWAY
+     */
+    private SqlSourceType rawScriptType = SqlSourceType.FILE;
+
+    /**
+     * default: ;
+     */
+    private String statementSeparator = ";";
+
+    /**
+     * print query result rows count in log, default 0 means do not print rows
+     */
+    private int maxPrintRows;
+
+    public boolean isSqlGateway() {
+        return sqlSubmitType == FlinkSqlSubmitType.SQL_GATEWAY && programType == ProgramType.SQL;
+    }
+
+    public Properties toJdbcProperties() {
+        Properties props = new Properties();
+        if (jdbcProperties != null) {
+            jdbcProperties.forEach((k, v) -> {
+                if (k != null && v != null) {
+                    props.put(k, v);
+                }
+            });
+        }
+        return props;
+    }
+
     @Override
     public boolean checkParameters() {
+        if (isSqlGateway()) {
+            return checkSqlGatewayParameters();
+        }
         /**
          * When saving a task, the parameter cannot be empty. There are two judgments:
          * (1) When ProgramType is SQL, rawScript cannot be empty.
@@ -109,11 +173,38 @@ public boolean checkParameters() {
         return programType != null && (rawScript != null || mainJar != null);
     }
 
+    private boolean checkSqlGatewayParameters() {
+        if (StringUtils.isBlank(flinkJdbcUrl)) {
+            return false;
+        }
+        if (initScriptType == SqlSourceType.FILE && isEmpty(initScriptResourceList)) {
+            return false;
+        }
+        if (rawScriptType == SqlSourceType.FILE) {
+            return !isEmpty(resourceList);
+        }
+        return rawScriptType == SqlSourceType.SCRIPT && StringUtils.isNotBlank(rawScript);
+    }
+
     @Override
     public List<ResourceInfo> getResourceFilesList() {
+        if (isSqlGateway()) {
+            List<ResourceInfo> list = new ArrayList<>();
+            if (initScriptType == SqlSourceType.FILE && !isEmpty(initScriptResourceList)) {
+                list.addAll(initScriptResourceList);
+            }
+            if (rawScriptType == SqlSourceType.FILE && !isEmpty(resourceList)) {
+                list.addAll(resourceList);
+            }
+            return list;
+        }
         if (mainJar != null && !resourceList.contains(mainJar)) {
             resourceList.add(mainJar);
         }
         return resourceList;
     }
+
+    private static boolean isEmpty(List<?> list) {
+        return list == null || list.isEmpty();
+    }
 }