This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12898-f3381ba41361cd39ec15dc40cd19239dd470e411 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c389c24b0ec24c55dafbf2d0e9d98189474d3e2f Author: zhiweiniu <[email protected]> AuthorDate: Tue Oct 27 12:09:05 2026 +0000 [Fix][Core] Fix application CLI validation and status handling (#12898) --- docs/en/getting-started/application/cli.md | 32 ++-- .../application/kubernetes/commands.md | 15 +- .../introduction/concepts/incompatible-changes.md | 6 + docs/zh/getting-started/application/cli.md | 29 +++- .../application/kubernetes/commands.md | 15 +- .../introduction/concepts/incompatible-changes.md | 6 + .../core/starter/utils/CommandLineUtils.java | 31 +++- .../starter/seatunnel/SeaTunnelApplication.java | 15 +- .../core/starter/seatunnel/SeaTunnelClient.java | 5 +- .../seatunnel/args/ApplicationCommandArgs.java | 54 +++++-- .../command/ApplicationExecuteCommand.java | 17 ++- .../starter/seatunnel/SeaTunnelClientOOMTest.java | 3 +- .../seatunnel/args/ClientCommandArgsTest.java | 23 +++ .../command/ApplicationExecuteCommandTest.java | 166 ++++++++++++++++++++- .../engine/common/runtime/ApplicationStatus.java | 4 +- 15 files changed, 362 insertions(+), 59 deletions(-) diff --git a/docs/en/getting-started/application/cli.md b/docs/en/getting-started/application/cli.md index cae235e005..cdcf9229c0 100644 --- a/docs/en/getting-started/application/cli.md +++ b/docs/en/getting-started/application/cli.md @@ -6,13 +6,13 @@ title: Command Line # Application command line `bin/seatunnel-application.sh` is the only entry point of Application Mode. It exposes three -operations, all of which target one external resource platform and read the same deployment file, +operations, all of which target one external resource platform using matching deployment options, so `status` and `cancel` see exactly the application `submit` created. ## Synopsis ```bash -bin/seatunnel-application.sh submit -t <target> -a application.config [-c job.conf] +bin/seatunnel-application.sh submit -t <target> [-a application.config] [-Dkey=value ...] [-c job.conf] bin/seatunnel-application.sh status -t <target> --id APPLICATION_ID -a application.config [--wait] \ [--wait-timeout <ms>] [--wait-interval <ms>] bin/seatunnel-application.sh cancel -t <target> --id APPLICATION_ID -a application.config @@ -57,7 +57,7 @@ Usage: seatunnel-application.sh [options] submit | status | cancel | `submit` / `status` / `cancel` | all | Operation, the first positional argument | | `-t`, `--target` | all | Resource platform that runs the application | | `-c`, `--config` | submit | SeaTunnel job configuration file; optional when the platform carries the job, such as a mounted `job.conf` | -| `-a`, `--application-config` | all | Optional deployment file with common and platform options | +| `-a`, `--application-config` | all | Optional deployment file with common and platform options; options can also be supplied through `-D` | | `-Dkey=value` | all | Override one non-sensitive `application.*` or `kubernetes.*` deployment option, for example `-Dapplication.worker-count=4`; repeatable. Job `${...}` variables are resolved inside the Pod, not on the submitting host | | `--id` | status, cancel | Platform application ID printed by `submit` | | `--wait` | submit, status | Poll until a terminal status | @@ -71,7 +71,7 @@ Usage: seatunnel-application.sh [options] submit | status | cancel | Operation | What it does | Result | | --- | --- | --- | | `submit` | Localizes the job, creates the application on the platform and starts it | Prints the application ID and the native job ID | -| `status` | Reads the platform state and maps it to an `ApplicationStatus` | Prints the application ID and status; exits non-zero when the application failed | +| `status` | Reads the platform state and maps it to an `ApplicationStatus` | Prints the application ID and status; a successful plain query exits `0`; with `--wait`, follows the [waiting exit codes](#status-and-waiting) | | `cancel` | Removes the application and the resources it owns | Prints the application ID and the cancellation result | ## Submit returns immediately @@ -89,6 +89,10 @@ The job comes from `-c` or from a file the platform mounts inside the applicatio resource is created. Values written as `${VAR}` are resolved inside the running application, never on the submitting host, so platform-managed credentials can be injected through the environment. +The deployment file (`-a`) is optional when `-D` supplies the required deployment options. +When both are given, `-D` overrides file values. Platform validation still applies: omitting +`-a` does not make incomplete deployment settings valid. + ## Status and waiting `status` reports one snapshot by default and polls that snapshot until it becomes terminal when @@ -106,14 +110,24 @@ often the status is polled. The platform-independent states are: | `CANCELED` | The platform reported a cancellation | | `UNKNOWN` | A known application has a platform state that cannot yet be mapped | -`SUCCEEDED`, `FAILED` and `CANCELED` are terminal; `--wait` stops there. `FAILED`, `CANCELED` and -`UNKNOWN` make `status` exit non-zero, which turns the command into a completion gate for -schedulers and CI: +`SUCCEEDED`, `FAILED` and `CANCELED` are terminal; `--wait` stops there. It also stops at +`UNKNOWN` when the platform cannot determine the application outcome. For both `status --wait` +and `submit --wait`, `FAILED` exits with code `1`, while `UNKNOWN` exits with code `2`: the result +is indeterminate, not a confirmed failure or success. `SUCCEEDED` and `CANCELED` still exit `0`. +A plain `status` prints the snapshot and exits `0`, including for `FAILED` and `UNKNOWN`; +query errors still exit non-zero. Do not interpret every non-zero exit as a failed application: ```bash -if ! bin/seatunnel-application.sh status -t <target> --id "$APP_ID" -a application.config --wait; then - echo "application $APP_ID did not succeed" +if bin/seatunnel-application.sh status -t <target> --id "$APP_ID" -a application.config --wait; then + result=0 +else + result=$? fi +case "$result" in + 0) echo "application status reported; inspect the printed status" ;; + 2) echo "application $APP_ID outcome is unknown" ;; + *) echo "application failed or the command could not complete (exit $result)" ;; +esac ``` ## Cancel diff --git a/docs/en/getting-started/application/kubernetes/commands.md b/docs/en/getting-started/application/kubernetes/commands.md index 378eae110b..29bcf1d16f 100644 --- a/docs/en/getting-started/application/kubernetes/commands.md +++ b/docs/en/getting-started/application/kubernetes/commands.md @@ -44,8 +44,11 @@ Application ID: example-9f2c41a0b7d3 Status: RUNNING ``` -`--wait` keeps polling until a terminal status is reported; `FAILED`, `CANCELED` and `UNKNOWN` make -the command exit non-zero, which turns it into a completion gate for schedulers and CI. +A plain `status` prints the snapshot and exits `0` when the query succeeds, including for `FAILED` +and `UNKNOWN`; query errors exit non-zero. `--wait` keeps polling until a terminal status or `UNKNOWN` +is reported. Both `status --wait` and `submit --wait` exit `1` for `FAILED`, `2` for `UNKNOWN` +(an indeterminate outcome, not a confirmed success or failure), and `0` for `SUCCEEDED` or `CANCELED`. +See [Status and waiting](../cli.md#status-and-waiting) for the shared exit-code contract. `--wait-timeout` bounds that polling and fails with the last observed status when the limit is reached, while the default `0` keeps waiting indefinitely; `--wait-interval` sets the poll interval (500 ms by default). @@ -54,10 +57,10 @@ reached, while the default `0` keeps waiting indefinitely; `--wait-interval` set `cancel` deletes the owner Job of the application. Owned Pods, Services, Secrets and ConfigMaps are garbage-collected through the owner reference; referenced objects created outside SeaTunnel are -never modified or deleted. After cancellation the application reports `CANCELED`, and a finished -Job is retained until `kubernetes.finished-job-retention-seconds` expires. A canceled master exits -with code 1, because a Kubernetes Job cannot express `CANCELED`, so the Job object ends in the -`Failed` state while `status` reports `CANCELED`. +never modified or deleted. The command reports `Cancellation requested` once the delete is accepted. +Because the owner Job is removed, a later `status` returns `UNKNOWN`; use `kubectl` to confirm that +the Pods are gone. If the delete call fails before the Job is removed, retry `cancel` after the API +outage is resolved. ## Inspecting in Kubernetes diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 29958c414f..3984ea546a 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -41,6 +41,12 @@ The same strictness applies wherever the shared JSON-to-row conversion is used w - **Impact**: With the defaults, timestamps are written as `yyyy-MM-dd HH:mm:ss` (a space instead of `T`, fractional seconds dropped) and times as `HH:mm:ss`. Readers that expect the previous ISO layout must be adjusted. - **Migration Guide**: To keep an ISO layout, set `datetime_format` on the sink, for example `YYYY_MM_DD_HH_MM_SS_ISO8601` (`yyyy-MM-dd'T'HH:mm:ss`) or `YYYY_MM_DD_HH_MM_SS_SSS_ISO8601` (`yyyy-MM-dd'T'HH:mm:ss.SSS`) to keep milliseconds, and read the files with the matching source `datetime_format`. +### Native CLI help exit code + +`seatunnel.sh -h` now exits with `0` instead of `234`. +Update scripts that recognize native help by exit code `234`. The Spark/Flink Java launcher +protocol still uses `234`; its wrapper scripts continue to translate help to exit `0`. + ### CDC: nullable columns are no longer used as the snapshot split column - **Behavior change: JDBC-based CDC sources (MySQL, PostgreSQL, Oracle, SQL Server, DB2) no longer split the snapshot on a nullable column** diff --git a/docs/zh/getting-started/application/cli.md b/docs/zh/getting-started/application/cli.md index da7433d3d7..1c19c11f4b 100644 --- a/docs/zh/getting-started/application/cli.md +++ b/docs/zh/getting-started/application/cli.md @@ -6,13 +6,13 @@ title: 命令行 # Application 命令行 `bin/seatunnel-application.sh` 是 Application Mode 的唯一入口，只提供三个操作。三者都面向同一个 -外部资源平台并读取同一份部署文件，因此 `status`、`cancel` 看到的正是 `submit` 创建的 +外部资源平台并使用一致的部署选项，因此 `status`、`cancel` 看到的正是 `submit` 创建的 application。 ## 命令格式 ```bash -bin/seatunnel-application.sh submit -t <target> -a application.config [-c job.conf] +bin/seatunnel-application.sh submit -t <target> [-a application.config] [-Dkey=value ...] [-c job.conf] bin/seatunnel-application.sh status -t <target> --id APPLICATION_ID -a application.config [--wait] \ [--wait-timeout <ms>] [--wait-interval <ms>] bin/seatunnel-application.sh cancel -t <target> --id APPLICATION_ID -a application.config @@ -57,7 +57,7 @@ Usage: seatunnel-application.sh [options] submit | status | cancel | `submit` / `status` / `cancel` | 全部 | 操作类型，位于第一个位置参数 | | `-t`、`--target` | 全部 | 运行 application 的资源平台 | | `-c`、`--config` | submit | SeaTunnel 作业配置文件；作业由平台提供（如挂载的 `job.conf`）时可省略 | -| `-a`、`--application-config` | 全部 | 可选部署文件，包含公共选项和平台选项 | +| `-a`、`--application-config` | 全部 | 包含公共选项和平台选项的可选部署文件；也可通过 `-D` 提供选项 | | `-Dkey=value` | 全部 | 覆盖单个非敏感的 `application.*` / `kubernetes.*` 部署选项（例如 `-Dapplication.worker-count=4`），可重复；作业 `${...}` 变量在 Pod 内解析，不在提交机替换 | | `--id` | status、cancel | `submit` 打印的平台 application ID | | `--wait` | submit、status | 轮询直到终态 | @@ -71,7 +71,7 @@ Usage: seatunnel-application.sh [options] submit | status | cancel | 操作 | 做什么 | 结果 | | --- | --- | --- | | `submit` | 完成作业本地化，在平台上创建 application 并启动 | 打印 application ID 与原生作业 ID | -| `status` | 读取平台状态并映射为 `ApplicationStatus` | 打印 application ID 与状态；application 失败时以非零码退出 | +| `status` | 读取平台状态并映射为 `ApplicationStatus` | 打印 application ID 与状态；普通查询成功时返回 `0`；带 `--wait` 时遵循[等待退出码规则](#状态与等待) | | `cancel` | 删除 application 及其自有资源 | 打印 application ID 与取消结果 | ## submit 立即返回 @@ -88,6 +88,9 @@ bin/seatunnel-application.sh submit -t <target> -a application.config --wait 的值在运行中的 application 内解析，不在提交机解析，因此可以通过环境变量注入平台管理的 凭据。 +通过 `-D` 提供所需部署选项时，可以省略部署文件（`-a`）。两者同时提供时，`-D` 覆盖文件值。 +平台配置校验仍然生效：省略 `-a` 不代表不完整的部署配置可以提交成功。 + ## 状态与等待 `status` 默认只读一次快照；加 `--wait` 后持续轮询直到终态。`--wait-timeout` 可以限制轮询时长， @@ -104,13 +107,23 @@ bin/seatunnel-application.sh submit -t <target> -a application.config --wait | `CANCELED` | 平台报告已取消 | | `UNKNOWN` | 已知 application 的平台状态暂时无法映射 | -`SUCCEEDED`、`FAILED`、`CANCELED` 为终态，`--wait` 在终态停止。`FAILED`、`CANCELED` 和 -`UNKNOWN` 会让 `status` 以非零码退出，因此可以直接作为调度器或 CI 的完成判定： +`SUCCEEDED`、`FAILED`、`CANCELED` 为终态，`--wait` 在终态停止；平台无法确定 application +结果时（`UNKNOWN`）也会停止。`status --wait` 和 `submit --wait` 在 `FAILED` 时退出码为 `1`， +在 `UNKNOWN` 时退出码为 `2`，表示无法确定成功或失败，而不是认定应用失败。 +`SUCCEEDED` 和 `CANCELED` 仍返回 `0`。不带 `--wait` 的 `status` 打印快照后返回 `0`， +包括 `FAILED` 和 `UNKNOWN`；查询异常仍以非零码退出。不要把所有非零退出码都当成应用失败： ```bash -if ! bin/seatunnel-application.sh status -t <target> --id "$APP_ID" -a application.config --wait; then - echo "application $APP_ID did not succeed" +if bin/seatunnel-application.sh status -t <target> --id "$APP_ID" -a application.config --wait; then + result=0 +else + result=$? fi +case "$result" in + 0) echo "application status reported; inspect the printed status" ;; + 2) echo "application $APP_ID outcome is unknown" ;; + *) echo "application failed or the command could not complete (exit $result)" ;; +esac ``` ## cancel diff --git a/docs/zh/getting-started/application/kubernetes/commands.md b/docs/zh/getting-started/application/kubernetes/commands.md index a0af9d9aea..2d47a9fb11 100644 --- a/docs/zh/getting-started/application/kubernetes/commands.md +++ b/docs/zh/getting-started/application/kubernetes/commands.md @@ -40,17 +40,20 @@ Application ID: example-9f2c41a0b7d3 Status: RUNNING ``` -加 `--wait` 会持续轮询直到终态；`FAILED`、`CANCELED` 和 `UNKNOWN` 会让命令以非零码退出，可直接 -作为调度器或 CI 的完成判定。`--wait-timeout` 可以限制轮询时长，超时后带最后状态报错退出； +不带 `--wait` 的 `status` 打印状态快照，查询成功时返回 `0`，包括 `FAILED` 和 `UNKNOWN`； +查询异常以非零码退出。加 `--wait` 会持续轮询直到终态或 `UNKNOWN`。 +`status --wait` 和 `submit --wait` 在 `FAILED` 时退出码为 `1`，在 `UNKNOWN` 时为 `2` +（表示无法确定成功或失败），在 `SUCCEEDED` 或 `CANCELED` 时为 `0`。 +共享退出码规则见[状态与等待](../cli.md#状态与等待)。 +`--wait-timeout` 可以限制轮询时长，超时后带最后状态报错退出； 默认 `0` 表示一直等待。`--wait-interval` 设置轮询间隔（默认 500 ms）。 ## cancel `cancel` 删除 application 的 owner Job，自有 Pod、Service、Secret 和 ConfigMap 通过 owner -reference 一并回收；SeaTunnel 之外创建的引用对象不会被修改或删除。取消后 application 状态为 -`CANCELED`，已完成的 Job 保留到 `kubernetes.finished-job-retention-seconds` 到期。 -被取消的 Master 以退出码 1 结束（Kubernetes Job 无法表达 `CANCELED`），因此 Job 对象最终是 -`Failed` 状态，而 `status` 报告 `CANCELED`。 +reference 一并回收；SeaTunnel 之外创建的引用对象不会被修改或删除。删除请求被接受后命令输出 +`Cancellation requested`。由于 owner Job 已被删除，之后再执行 `status` 会返回 `UNKNOWN`； +可以用 `kubectl` 确认 Pod 已经消失。如果删除前 API 调用失败，请在 API 恢复后重试 `cancel`。 ## 用 kubectl 排查 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index f00ffe9c47..3a654648af 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -40,6 +40,12 @@ - **影响**：使用默认配置时，时间戳写成 `yyyy-MM-dd HH:mm:ss`（`T` 改为空格，丢弃小数秒），时间写成 `HH:mm:ss`。依赖原 ISO 格式的下游需要调整。 - **迁移指南**：如需保留 ISO 格式，请在 Sink 上配置 `datetime_format`，例如 `YYYY_MM_DD_HH_MM_SS_ISO8601`（`yyyy-MM-dd'T'HH:mm:ss`），或使用 `YYYY_MM_DD_HH_MM_SS_SSS_ISO8601`（`yyyy-MM-dd'T'HH:mm:ss.SSS`）保留毫秒，并在读取时为 Source 配置对应的 `datetime_format`。 +### 原生 CLI 帮助退出码 + +`seatunnel.sh -h` 现在以 `0` 而非 `234` 退出。 +依赖退出码 `234` 识别原生帮助的脚本需要调整。Spark/Flink Java 启动器仍使用 `234` 协议， +对应包装脚本仍将帮助请求转换为退出码 `0`。 + ### CDC：可空列不再用作快照分片列 - **行为变更：基于 JDBC 的 CDC 源（MySQL、PostgreSQL、Oracle、SQL Server、DB2）不再使用可空列切分快照** diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/CommandLineUtils.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/CommandLineUtils.java index 01d82302fc..9b884e660f 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/CommandLineUtils.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/CommandLineUtils.java @@ -55,6 +55,22 @@ public class CommandLineUtils { String programName, boolean acceptUnknownOptions, Consumer<T> commandValidator) { + return parse( + args, obj, programName, acceptUnknownOptions, commandValidator, USAGE_EXIT_CODE); + } + + /** + * Parses arguments with an explicit help exit code. The default overloads retain the usage exit + * code expected by the Spark and Flink launchers; direct entrypoints may opt in to a zero exit + * code for help. Invalid arguments always use the usage exit code. + */ + public static <T extends CommandArgs> T parse( + String[] args, + T obj, + String programName, + boolean acceptUnknownOptions, + Consumer<T> commandValidator, + int helpExitCode) { List<String> list = Arrays.asList(args); if (list.contains("-can") || list.contains("--cancel") @@ -76,21 +92,22 @@ public class CommandLineUtils { jCommander.parse(args); // The args is not belongs to SeaTunnel, add into engine original parameters obj.setOriginalParameters(jCommander.getUnknownOptions()); + // Help must short-circuit before command validation: a bare --help has no + // operation/id, and validating it would report an unrelated argument error. + if (obj.isHelp()) { + exit(jCommander, helpExitCode); + } commandValidator.accept(obj); } catch (ParameterException e) { System.err.println(e.getLocalizedMessage()); - exit(jCommander); - } - - if (obj.isHelp()) { - exit(jCommander); + exit(jCommander, USAGE_EXIT_CODE); } return obj; } - private static void exit(JCommander jCommander) { + private static void exit(JCommander jCommander, int exitCode) { jCommander.setUsageFormatter(new UsageFormatter(jCommander)); jCommander.usage(); - System.exit(USAGE_EXIT_CODE); + System.exit(exitCode); } } diff --git a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelApplication.java b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelApplication.java index 76f4f37e98..a736a97f05 100644 --- a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelApplication.java +++ b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelApplication.java @@ -20,10 +20,12 @@ package org.apache.seatunnel.core.starter.seatunnel; import org.apache.seatunnel.common.constants.EngineType; import org.apache.seatunnel.core.starter.SeaTunnel; import org.apache.seatunnel.core.starter.seatunnel.args.ApplicationCommandArgs; +import org.apache.seatunnel.core.starter.seatunnel.command.ApplicationExecuteCommand; import org.apache.seatunnel.core.starter.utils.CommandLineUtils; /** User-facing entrypoint for native Zeta application deployment and management. */ public final class SeaTunnelApplication { + private static final int HELP_EXIT_CODE = 0; public static void main(String[] args) { ApplicationCommandArgs applicationCommandArgs = @@ -32,13 +34,22 @@ public final class SeaTunnelApplication { new ApplicationCommandArgs(), EngineType.SEATUNNEL_APPLICATION.getStarterShellName(), false, - ApplicationCommandArgs::validateCommandOptions); + ApplicationCommandArgs::validateCommandOptions, + HELP_EXIT_CODE); + final int exitCode; try { - SeaTunnel.run(applicationCommandArgs.buildCommand()); + ApplicationExecuteCommand command = + (ApplicationExecuteCommand) applicationCommandArgs.buildCommand(); + SeaTunnel.run(command); + exitCode = command.getExitCode(); } catch (Exception e) { System.err.println("Application command failed: " + e.getMessage()); e.printStackTrace(System.err); System.exit(1); + return; + } + if (exitCode != 0) { + System.exit(exitCode); } } } diff --git a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClient.java b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClient.java index e9972e8f34..0213d49e2c 100644 --- a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClient.java +++ b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClient.java @@ -29,6 +29,8 @@ import lombok.extern.slf4j.Slf4j; @Slf4j public class SeaTunnelClient { + private static final int HELP_EXIT_CODE = 0; + public static void main(String[] args) throws CommandException { ClientCommandArgs clientCommandArgs = CommandLineUtils.parse( @@ -36,7 +38,8 @@ public class SeaTunnelClient { new ClientCommandArgs(), EngineType.SEATUNNEL.getStarterShellName(), true, - ClientCommandArgs::validateCommandOptions); + ClientCommandArgs::validateCommandOptions, + HELP_EXIT_CODE); try { SeaTunnel.run(clientCommandArgs.buildCommand()); } catch (Error e) { diff --git a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ApplicationCommandArgs.java b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ApplicationCommandArgs.java index 16c5b6dcd9..472a0f742f 100644 --- a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ApplicationCommandArgs.java +++ b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ApplicationCommandArgs.java @@ -118,24 +118,32 @@ public class ApplicationCommandArgs extends CommandArgs { if (applicationConfig != null) { requireFile(applicationConfig, "--application-config"); } - if (waitTimeoutMillis != null && waitTimeoutMillis < 0) { - throw new ParameterException("--wait-timeout must not be negative"); - } - if (waitIntervalMillis != null && waitIntervalMillis < 1) { - throw new ParameterException("--wait-interval must be positive"); + validateWaitOptions(); + if (operation == ApplicationOperation.SUBMIT) { + validateSubmitOptions(); + } else { + validateStatusOrCancelOptions(); } - if (!wait && (waitTimeoutMillis != null || waitIntervalMillis != null)) { - throw new ParameterException("--wait-timeout and --wait-interval require --wait"); + } + + /** + * Checks an explicitly supplied job file and rejects an application ID for submission. + * Deployment configuration remains optional here so complete {@code -D} options can supply it. + */ + private void validateSubmitOptions() { + if (config != null) { + requireFile(config, "--config"); } - if (operation == ApplicationOperation.SUBMIT) { - if (config != null) { - requireFile(config, "--config"); - } - if (id != null) { - throw new ParameterException("submit does not accept --id"); - } - return; + if (id != null) { + throw new ParameterException("submit does not accept --id"); } + } + + /** + * Requires a nonblank application ID and rejects submit-only options for status and cancel. + * Only status may wait for completion; cancel does not accept {@code --wait}. + */ + private void validateStatusOrCancelOptions() { if (id == null || id.trim().isEmpty()) { throw new ParameterException("status/cancel require --id"); } @@ -151,6 +159,22 @@ public class ApplicationCommandArgs extends CommandArgs { } } + /** + * Requires a nonnegative timeout and a positive poll interval when explicitly supplied. Both + * timing options require {@code --wait}; a zero timeout allows indefinite waiting. + */ + private void validateWaitOptions() { + if (waitTimeoutMillis != null && waitTimeoutMillis < 0) { + throw new ParameterException("--wait-timeout must not be negative"); + } + if (waitIntervalMillis != null && waitIntervalMillis < 1) { + throw new ParameterException("--wait-interval must be positive"); + } + if (!wait && (waitTimeoutMillis != null || waitIntervalMillis != null)) { + throw new ParameterException("--wait-timeout and --wait-interval require --wait"); + } + } + /** @return configured wait timeout in milliseconds, or {@code 0} to wait indefinitely */ public long resolvedWaitTimeoutMillis() { return waitTimeoutMillis == null ? 0L : waitTimeoutMillis; diff --git a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommand.java b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommand.java index dfdebb28d7..025a3adfaa 100644 --- a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommand.java +++ b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommand.java @@ -40,7 +40,16 @@ import java.util.concurrent.TimeUnit; */ public class ApplicationExecuteCommand implements Command<ApplicationCommandArgs> { + /** The platform cannot determine whether the application succeeded or failed. */ + public static final int UNKNOWN_STATUS_EXIT_CODE = 2; + private final ApplicationCommandArgs applicationCommandArgs; + private int exitCode; + + /** Returns the CLI outcome after execution without changing the reported application status. */ + public int getExitCode() { + return exitCode; + } public ApplicationExecuteCommand(ApplicationCommandArgs applicationCommandArgs) { this.applicationCommandArgs = applicationCommandArgs; @@ -48,6 +57,7 @@ public class ApplicationExecuteCommand implements Command<ApplicationCommandArgs @Override public void execute() throws CommandExecuteException { + exitCode = 0; try { Map<String, String> options = resolveOptions(); ClusterClientServiceLoader clientServiceLoader = new ClusterClientServiceLoader(); @@ -98,7 +108,12 @@ public class ApplicationExecuteCommand implements Command<ApplicationCommandArgs try (ClusterDescriptor<ID> descriptor = factory.create(options)) { ApplicationStatus status = getStatus(descriptor, applicationId); System.out.println("Status: " + status); - if (status.isFailure()) { + if (applicationCommandArgs.isWait() && status == ApplicationStatus.UNKNOWN) { + exitCode = UNKNOWN_STATUS_EXIT_CODE; + System.err.println( + "Application outcome is unknown; success or failure cannot be determined."); + } + if (applicationCommandArgs.isWait() && status.isFailure()) { throw new CommandExecuteException( "Application finished with an unsuccessful status"); } diff --git a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClientOOMTest.java b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClientOOMTest.java index 6e348a59f6..024c1ec0a2 100644 --- a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClientOOMTest.java +++ b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelClientOOMTest.java @@ -50,7 +50,8 @@ public class SeaTunnelClientOOMTest { Mockito.any(ClientCommandArgs.class), Mockito.anyString(), Mockito.anyBoolean(), - Mockito.any())) + Mockito.any(), + Mockito.eq(0))) .thenReturn(clientCommandArgs); // Mock SeaTunnel.run to throw OutOfMemoryError diff --git a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java index 2d1bb9d9da..3e1fd3504c 100644 --- a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java +++ b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java @@ -21,6 +21,7 @@ import org.apache.seatunnel.core.starter.SeaTunnel; import org.apache.seatunnel.core.starter.enums.DryRun; import org.apache.seatunnel.core.starter.enums.MasterType; import org.apache.seatunnel.core.starter.exception.CommandExecuteException; +import org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient; import org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand; import org.apache.seatunnel.core.starter.seatunnel.multitable.MultiTableSinkTest; import org.apache.seatunnel.core.starter.utils.CommandLineUtils; @@ -168,6 +169,28 @@ public class ClientCommandArgsTest { Assertions.assertEquals(USAGE_EXIT_CODE, statusCode); } + @Test + public void testHelpExitsZeroBeforeValidation() throws Exception { + int statusCode = catchSystemExit(() -> SeaTunnelClient.main(new String[] {"--help"})); + + Assertions.assertEquals(0, statusCode); + } + + @Test + public void testDefaultHelpRetainsUsageExitCodeBeforeValidation() throws Exception { + int statusCode = + catchSystemExit( + () -> + CommandLineUtils.parse( + new String[] {"--help"}, + new ClientCommandArgs(), + "legacy-starter", + true, + ignored -> Assertions.fail("Help must skip validation"))); + + Assertions.assertEquals(USAGE_EXIT_CODE, statusCode); + } + @Test public void testSampleDryRunRejectsJobControlOptions() { String[] args = {"-c", "app.conf", "--master", "local", "--dry-run", "sample", "--check"}; diff --git a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommandTest.java b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommandTest.java index 5478b3367f..0a0dec0b56 100644 --- a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommandTest.java +++ b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ApplicationExecuteCommandTest.java @@ -21,7 +21,9 @@ import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; import org.apache.seatunnel.common.constants.ApplicationOperation; import org.apache.seatunnel.core.starter.exception.CommandExecuteException; +import org.apache.seatunnel.core.starter.seatunnel.SeaTunnelApplication; import org.apache.seatunnel.core.starter.seatunnel.args.ApplicationCommandArgs; +import org.apache.seatunnel.core.starter.utils.CommandLineUtils; import org.apache.seatunnel.engine.client.deployment.ApplicationClusterDeployer; import org.apache.seatunnel.engine.client.deployment.ApplicationClusterDescriptorFactory; import org.apache.seatunnel.engine.client.deployment.ClusterClientServiceLoader; @@ -49,12 +51,15 @@ import java.util.Collections; import java.util.Map; import java.util.ServiceLoader; +import static com.github.stefanbirkner.systemlambda.SystemLambda.catchSystemExit; +import static org.apache.seatunnel.core.starter.constants.SeaTunnelStarterConstants.USAGE_EXIT_CODE; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockConstruction; @@ -310,6 +315,121 @@ class ApplicationExecuteCommandTest { verify(descriptor, never()).retrieve(any()); } + @Test + void statusWithoutWaitReportsUnknownWithoutFailing() throws Exception { + assertEquals(0, executeStatus(ApplicationStatus.UNKNOWN, false)); + } + + @Test + void statusWithWaitReportsUnknownWithDistinctExitCode() throws Exception { + assertEquals( + ApplicationExecuteCommand.UNKNOWN_STATUS_EXIT_CODE, + executeStatus(ApplicationStatus.UNKNOWN, true)); + } + + @Test + void statusWithWaitReportsCanceledWithoutFailing() throws Exception { + assertEquals(0, executeStatus(ApplicationStatus.CANCELED, true)); + } + + @Test + void statusWithoutWaitReportsFailedWithoutFailing() throws Exception { + assertEquals(0, executeStatus(ApplicationStatus.FAILED, false)); + } + + @Test + void statusWithWaitFailsOnFailed() { + assertThrows( + CommandExecuteException.class, () -> executeStatus(ApplicationStatus.FAILED, true)); + } + + @ParameterizedTest + @ValueSource(strings = {"status", "submit"}) + void unknownWaitExitsTwoFromApplicationEntrypoint(String operation) throws Exception { + Path application = temporary.resolve("application.config"); + Files.write(application, "application.name = demo".getBytes(StandardCharsets.UTF_8)); + ClusterDescriptor<String> descriptor = mock(ClusterDescriptor.class); + ApplicationClusterDescriptorFactory<String> factory = + mock(ApplicationClusterDescriptorFactory.class); + when(factory.create(any())).thenReturn(descriptor); + when(factory.parseApplicationId("application_1")).thenReturn("native-id"); + when(descriptor.deployApplication(any())).thenReturn("application_1"); + when(descriptor.getApplicationStatus("native-id")).thenReturn(ApplicationStatus.UNKNOWN); + String[] arguments = + "submit".equals(operation) + ? new String[] { + "submit", "-t", "kubernetes", "-a", application.toString(), "--wait" + } + : new String[] { + "status", "-t", "kubernetes", "--id", "application_1", "--wait" + }; + try (MockedConstruction<ClusterClientServiceLoader> loaders = + mockConstruction( + ClusterClientServiceLoader.class, + (loader, context) -> + when(loader.<String>getClusterClientFactory(DeployType.KUBERNETES)) + .thenReturn(factory))) { + assertEquals(2, catchSystemExit(() -> SeaTunnelApplication.main(arguments))); + } + verify(descriptor).getApplicationStatus("native-id"); + verify(descriptor, times("submit".equals(operation) ? 2 : 1)).close(); + verify(descriptor, never()).cancelApplication(any()); + } + + @Test + void helpExitsZeroBeforeCommandValidation() throws Exception { + int statusCode = + catchSystemExit( + () -> + SeaTunnelApplication.main( + new String[] {"status", "-t", "kubernetes", "--help"})); + assertEquals(0, statusCode); + } + + @Test + void submitWithMissingApplicationConfigExitsThroughUsagePath() throws Exception { + int statusCode = + catchSystemExit( + () -> + CommandLineUtils.parse( + new String[] { + "submit", + "-t", + "kubernetes", + "-a", + temporary.resolve("missing.config").toString() + }, + new ApplicationCommandArgs(), + "seatunnel-application.sh", + false, + ApplicationCommandArgs::validateCommandOptions, + 0)); + assertEquals(USAGE_EXIT_CODE, statusCode); + } + + private int executeStatus(ApplicationStatus status, boolean wait) throws Exception { + ClusterDescriptor<String> descriptor = mock(ClusterDescriptor.class); + ApplicationClusterDescriptorFactory<String> factory = + mock(ApplicationClusterDescriptorFactory.class); + when(factory.parseApplicationId("application_1")).thenReturn("native-id"); + doReturn(descriptor).when(factory).create(any()); + when(descriptor.getApplicationStatus("native-id")).thenReturn(status); + try (MockedConstruction<ClusterClientServiceLoader> loaders = + mockConstruction( + ClusterClientServiceLoader.class, + (loader, context) -> + when(loader.<String>getClusterClientFactory(DeployType.KUBERNETES)) + .thenReturn(factory))) { + ApplicationCommandArgs args = + parse("status", "-t", "kubernetes", "--id", "application_1"); + args.setWait(wait); + ApplicationExecuteCommand command = (ApplicationExecuteCommand) args.buildCommand(); + command.execute(); + verify(descriptor).close(); + return command.getExitCode(); + } + } + @Test void waitsForPlatformTerminalState() throws Exception { ClusterDescriptor<String> descriptor = mock(ClusterDescriptor.class); @@ -376,7 +496,7 @@ class ApplicationExecuteCommandTest { ApplicationClusterDescriptorFactory<String> factory = mock(ApplicationClusterDescriptorFactory.class); when(factory.parseApplicationId("application_1")).thenReturn("native-id"); - org.mockito.Mockito.doReturn(descriptor).when(factory).create(any()); + doReturn(descriptor).when(factory).create(any()); when(descriptor.getApplicationStatus("native-id")).thenReturn(ApplicationStatus.SUCCEEDED); try (MockedConstruction<ClusterClientServiceLoader> loaders = mockConstruction( @@ -405,6 +525,50 @@ class ApplicationExecuteCommandTest { verify(descriptor, times(2)).close(); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void submitsWithDeploymentOverridesOnly(boolean wait) throws Exception { + ClusterDescriptor<String> descriptor = mock(ClusterDescriptor.class); + ApplicationClusterDescriptorFactory<String> factory = + mock(ApplicationClusterDescriptorFactory.class); + when(factory.create(any())).thenReturn(descriptor); + when(descriptor.deployApplication(any())).thenReturn("application_1"); + when(factory.parseApplicationId("application_1")).thenReturn("native-id"); + when(descriptor.getApplicationStatus("native-id")).thenReturn(ApplicationStatus.SUCCEEDED); + try (MockedConstruction<ClusterClientServiceLoader> loaders = + mockConstruction( + ClusterClientServiceLoader.class, + (loader, context) -> + when(loader.<String>getClusterClientFactory(DeployType.KUBERNETES)) + .thenReturn(factory))) { + ApplicationCommandArgs args = + parse( + "submit", + "-t", + "kubernetes", + "-Dapplication.name=overrides-only", + "-Dapplication.worker-count=3", + "-Dkubernetes.image=example:latest", + "-Dkubernetes.job-config-file=/etc/seatunnel/job.conf"); + args.setWait(wait); + args.buildCommand().execute(); + } + ArgumentCaptor<ApplicationSpecification> specification = + ArgumentCaptor.forClass(ApplicationSpecification.class); + verify(descriptor).deployApplication(specification.capture()); + assertEquals("overrides-only", specification.getValue().getName()); + assertEquals(3, specification.getValue().getWorkerCount()); + assertEquals("", specification.getValue().getJobConfig()); + ArgumentCaptor<Map<String, String>> deploymentOptions = ArgumentCaptor.forClass(Map.class); + verify(factory, times(wait ? 2 : 1)).create(deploymentOptions.capture()); + assertEquals("example:latest", deploymentOptions.getValue().get("kubernetes.image")); + assertEquals( + "/etc/seatunnel/job.conf", + deploymentOptions.getValue().get("kubernetes.job-config-file")); + verify(descriptor, times(wait ? 1 : 0)).getApplicationStatus("native-id"); + verify(descriptor, times(wait ? 2 : 1)).close(); + } + @Test void validatesCommandBeforeLoadingAnyPlatformProvider() throws Exception { Path deploymentConfig = temporary.resolve("deployment.conf"); diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/runtime/ApplicationStatus.java b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/runtime/ApplicationStatus.java index 80b43607fb..0193e524db 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/runtime/ApplicationStatus.java +++ b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/runtime/ApplicationStatus.java @@ -39,8 +39,8 @@ public enum ApplicationStatus { return this == SUCCEEDED || this == FAILED || this == CANCELED; } - /** @return whether this state represents a failed, canceled, or unknown application */ + /** @return whether this state represents a failed application */ public boolean isFailure() { - return this == FAILED || this == CANCELED || this == UNKNOWN; + return this == FAILED; } }
