Repository navigation
Conversation
Keep sql-client.sh when sqlSubmitType is absent or CLIENT. SQL_GATEWAY submits over JDBC and cancels the statement.
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
- Wait for the submitted Flink job before reporting success
FlinkSqlGatewayExecutor.java:152–164, handle():84
Flink JDBC returns a JobID result for INSERT once the job is submitted. Consuming that result does not wait for job completion. This executor discards the JobID and reports SUCCESS, allowing downstream tasks to run before the batch finishes and hiding subsequent Flink job failures. Track the submitted job through its terminal state and propagate failures. Please cover a job that fails after successful submission.
- Preserve cancellation during connection establishment
FlinkSqlGatewayExecutor.java:76–84, cancel():98–108
If cancellation happens while DriverManager.getConnection() is blocked, both fields can still be null, so cancel() only sets EXIT_CODE_KILL. When connection establishment returns, handle() creates the statement, executes the scripts, and overwrites KILL with SUCCESS. Coordinate cancellation with resource publication, check cancellation before submission, and prevent success from overwriting a cancelled state.
- Split SQL without breaking quoted strings and comments
FlinkSqlGatewayExecutor.java:172–175
String.split() treats every separator occurrence as a statement boundary. With the default separator, valid SQL such as SELECT 'a;b'; becomes two invalid statements. Separators inside comments also corrupt the script. Use SQL-aware splitting and cover quoted literals, escaped quotes, and comments containing the separator.
- Clear or reject Gateway mode when switching away from SQL
format-data.ts:93–109; FlinkParameters.isSqlGateway()
After selecting SQL_GATEWAY, changing programType to JAVA/SCALA/PYTHON hides the Gateway controls but retains sqlSubmitType. Serialization still writes SQL_GATEWAY, and the backend branches solely on that field, so it executes the retained SQL instead of the newly configured application. Restrict Gateway mode to programType=SQL in serialization and backend validation, and reset incompatible state when the program type changes.
- Hide unsupported Gateway controls for FLINK_STREAM
use-flink-sql-gateway.ts:153–166
FLINK_STREAM also uses Fields.useFlink(model), so selecting SQL exposes the new Gateway selector and fields. However, formatParams() serializes Gateway settings only for FLINK, and FlinkStreamTask continues through the CLI path. A user can select Gateway mode and still get sql-client.sh execution. Gate these controls on taskType=FLINK and cover the shared FLINK_STREAM form.
- Resolve the JDBC driver's Java 8 incompatibility
dolphinscheduler-task-flink/pom.xml:50–54
The published flink-sql-jdbc-driver:2.2.0 contains Java 11 bytecode (class-file version 55), while this project still targets Java 8 and documents JDK 8 support. SQL_GATEWAY therefore cannot load this driver on a supported JDK 8 worker. Choose a compatible implementation or explicitly resolve the runtime compatibility requirement before introducing this dependency.
The executor issues SET 'table.dml-sync'='true' itself before user SQL, so a batch job is success only after it finishes and a later failure fails the task. Gateway mode is limited to FLINK SQL, statement splitting skips quotes and comments, and JDK 8 is rejected.
This branch has not been deployed
2 participants
Was this PR generated or assisted by AI?
YES — the SQL Gateway submit path, UI fields, and unit tests were assisted by an AI agent (Hermes). All changes were human-reviewed and verified by build + unit tests plus a smoke run on an internal cluster.
Purpose of the pull request
close #18691
Add an optional SQL Gateway submit mode to the existing FLINK task, so batch SQL can be sent to a remote Flink SQL Gateway through its JDBC driver (
flink-sql-jdbc-driver), enabling DolphinScheduler to schedule batch SQL workloads without a local Flink client installation.Today the
FLINKtask runs SQL through${FLINK_HOME}/bin/sql-client.sh, which requires a full Flink distribution on every worker host. Deployments that run a dedicated Flink SQL Gateway cluster instead can submit scripts straight to the Gateway over JDBC.Compared with spawning
sql-client.shon the worker, submitting through the SQL Gateway costs less memory on the worker host (no local Flink CLI/mini-cluster JVM per task), produces clearer errors (statement-level JDBC exceptions instead of parsing shell/YARN logs), and is faster (no CLI process startup and script materialization per statement).The new mode is opt-in via a
sqlSubmitTypeparameter: absent orCLIENTkeeps the existingsql-client.shpath unchanged;SQL_GATEWAYsubmits over JDBC.Brief change log
FlinkParameters.sqlSubmitType: absent /CLIENTkeeps${FLINK_HOME}/bin/sql-client.sh;SQL_GATEWAYsubmits over JDBC and cancels viastatement.cancel()FlinkTask.handle()/cancelApplication().getScript()andFlinkArgsUtilsare unchanged, so existing CLIENT behavior and FLINK_STREAM are untouchedorg.apache.dolphinscheduler.plugin.task.api.enums.SqlSourceType(SCRIPT/FILE, introduced by [Improvement-18019][task-sql] Support SQL from resource file and parameter placeholders #18020) for both the init script and the main script, and the JDBC URL and script content resolve parameter placeholders (e.g.${var}) before submission; statement splitting and leading-comment stripping are package-private static methods covered by plain JUnit tests (no Mockito);); optionalmaxPrintRowscaps query-result row logging;flinkJdbcUrlsupports parameter placeholders; optionaljdbcPropertiesmap filters null keys/values before connectdolphinscheduler-task-flink;dolphinscheduler-task-flink-streamexcludes it so its shade jar stays free of the driverSQLand submit type is SQL Gateway;use-resourcesfield helper gains optionalfield/limitparameters to support the init-script resource field; locales (en/zh)Verify this pull request
This change added tests and can be verified as follows:
mvn test -pl dolphinscheduler-task-plugin/dolphinscheduler-task-flink -Dtest=FlinkParametersTest,FlinkSqlGatewayExecutorTest,FlinkArgsUtilsTest,FlinkTaskTest -Dspotless.skip=true -Djacoco.skip=truemvn -pl dolphinscheduler-task-plugin/dolphinscheduler-task-flink,dolphinscheduler-task-plugin/dolphinscheduler-task-flink-stream spotless:checkFLINKtask withsqlSubmitType=SQL_GATEWAYwas executed on an internal cluster (worker-only smoke): the task instance ranFlinkSqlGatewayExecutor, connected tojdbc:flink://...:8083, executed init + main scripts from resource-center files, and cancelled viastatement.cancel()