JeremyXin commented on PR #11421:
URL: https://github.com/apache/seatunnel/pull/11421#issuecomment-4980301350
> Thanks for working on this. I re-reviewed the latest head `8af33cf91235`
from scratch.
>
> # What this PR fixes
> * User pain: When a Zeta streaming job is cancelled or fails, users cannot
resume it from its last successful checkpoint — the only recovery path is
savepoint-based (`--restore`), and cancelled jobs eagerly clean up checkpoint
data, so long-running CDC pipelines lose incremental state (#10926).
> * Fix approach: Adds a `CHECKPOINT` restore mode wired end-to-end — a new
`--restore-with-checkpoint <job-id>` CLI flag and
`restoreMode=CHECKPOINT&restoreSourceJobId=` REST parameter that route through
`EngineClient.restoreFromCheckpointExecutionContext(...)`, plus a new
`checkpoint.retain-after-job-cancelled` env option (with cluster-level default
in `seatunnel.yaml`) so cancelled jobs can keep their checkpoint data; docs
updated in EN/ZH.
> * One-line summary: The feature is coherently plumbed from CLI/REST into
job submission with mutual-exclusion validation and CLI tests, but there are
open concerns before merge: a possible public API change in
`SeaTunnelGetJobCheckpointCodec`, no authorization check on
`restoreSourceJobId` in the REST path, raw `Long.parseLong` on the
trimmed-but-unparsed checkpoint job id (NumberFormatException on non-numeric
input), `--savepoint` + `--restore-with-checkpoint` being silently accepted,
unbounded retention of serialized state when `retain-after-job-cancelled` is
enabled, and a docs inconsistency between the
`checkpoint.retain-after-job-cancelled` env key and the bare
`retain-after-job-cancelled` key shown in state-storage-and-recovery.md.
>
> # Runtime chain I rechecked
> ```
> CLI restore-from-checkpoint path (the changed code's primary runtime path):
>
> seatunnel.sh --restore-with-checkpoint <job-id>
> -> CommandLineUtils.parse(...) populates restoreWithCheckpointJobId
ClientCommandArgs.java:65-69
> -> ClientCommandArgs.buildCommand() ClientCommandArgs.java:154-162
> - rejects --restore + --restore-with-checkpoint together (line 155)
> - rejects blank restoreWithCheckpointJobId via trim().isEmpty()
(line 159) [note: does NOT validate numeric format, and does NOT check
--savepoint combination]
> -> ClientExecuteCommand.execute() ClientExecuteCommand.java:194-205
> - branch: null != clientCommandArgs.getRestoreWithCheckpointJobId()
> - Long.parseLong(clientCommandArgs.getRestoreWithCheckpointJobId())
ClientExecuteCommand.java:201 [raw NumberFormatException risk; input not
trimmed here]
> - optional custom jobId:
Long.parseLong(clientCommandArgs.getCustomJobId())
ClientExecuteCommand.java:203
> -> EngineClient.restoreFromCheckpointExecutionContext(configFile,
variables, jobConfig, seaTunnelConfig, restoreSourceJobId, customJobId)
EngineClient.java (new entrypoint, not in shown diff)
> -> job submission -> checkpoint lookup via
SeaTunnelGetJobCheckpointCodec SeaTunnelGetJobCheckpointCodec.java [flagged
public API change]
> -> CheckpointCoordinator/state restore on the server side, gated by
checkpoint retention:
> EnvCommonOptions.CHECKPOINT_RETAIN_AFTER_JOB_CANCELLED
EnvCommonOptions.java:104-110 (job-level env, overrides cluster default)
> registered in EnvOptionRule.optionRule() EnvOptionRule.java:46
>
> Parallel REST path (documented, server code outside shown diff):
> POST /submit-job?restoreMode=CHECKPOINT&restoreSourceJobId=<id>
rest-api-job-lifecycle.md:249-262
> -> same restore-from-checkpoint submission flow on the master node [no
caller ownership/authz check evident on restoreSourceJobId]
> ```
>
> # Findings
> **Issue 1: Possible public API change in
seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/protocol/codec/SeaTunnelGetJobCheckpointCodec.java**
>
> * Location:
`seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/protocol/codec/SeaTunnelGetJobCheckpointCodec.java`
> * Why it matters: A public/protected signature appears to be removed or
changed. Verify backward compatibility for downstream connectors.
> * Severity: High
>
> **Issue 2: REST restore path lets any caller restore an arbitrary
historical job's checkpoint state via restoreSourceJobId with no
ownership/authorization check evident**
>
> * Location: `docs/en/engines/zeta/rest-api-job-lifecycle.md:253`
> * Why it matters: - Root cause: the new REST contract documented in this
PR accepts `restoreMode=CHECKPOINT&restoreSourceJobId=<any jobId>` as plain
query parameters on `/submit-job`, and nothing in the documented flow ties the
restore-source job to the caller's identity — any client that can reach the
REST port can attach any historical job's checkpoint state to a brand-new job
graph of their own design.
> * Attack scenario / blast radius: in a shared Zeta cluster, an attacker
who knows or enumerates job IDs (they are visible via existing job-list
endpoints) can submit a job whose source config mimics the victim's (e.g. a CDC
source) with `restoreSourceJobId` pointing at the victim job, hijacking the
victim's binlog offsets/enumerator state, then routing the replayed data to a
sink the attacker controls. Checkpoint state can embed connection offsets and
serialized reader/enumerator internals, so this is cross-tenant state
disclosure and data-flow hijacking, not just a resume convenience.
> * Fix: gate the CHECKPOINT/SAVEPOINT restore modes behind the existing
REST API authentication, and on the server side validate that
`restoreSourceJobId` refers to an existing job whose checkpoint the caller is
allowed to reuse (at minimum log an audit record including caller identity and
source jobId). The docs should also state explicitly that this endpoint must
not be exposed on untrusted networks.
> * Evidence: `docs/en/engines/zeta/rest-api-job-lifecycle.md:305 curl -X
POST
"http://<master>:8080/submit-job?restoreMode=SAVEPOINT&restoreSourceJobId=733584788375093248"
\`
> * Suggested fix: This new REST contract lets any caller who can reach
`/submit-job` attach an arbitrary historical job's checkpoint state via
`restoreSourceJobId`. On a shared cluster, job IDs are enumerable through the
list endpoints, so an attacker can hijack another job's checkpoint state (e.g.
CDC offsets and serialized enumerator internals) and replay its data into a
sink they control. Please make sure the server-side handler enforces the
existing REST authentication for restore modes and validates/audits the
`restoreSourceJobId` (caller identity + source job), and document here that
this endpoint must not be exposed on untrusted networks.
> * Severity: Medium
>
> **Issue 3: Non-numeric or whitespace-padded --restore-with-checkpoint job
id passes buildCommand() validation but crashes later with a raw
NumberFormatException**
>
> * Location:
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgs.java:159`
> * Why it matters: - Root cause: `buildCommand()` only rejects a job id
whose `trim()` is empty, but `ClientExecuteCommand.execute()` calls
`Long.parseLong(clientCommandArgs.getRestoreWithCheckpointJobId())` on the
untrimmed raw string. A value like `" 123 "` survives the validation
(`trim().isEmpty()` is false) and then `Long.parseLong(" 123 ")` throws
`NumberFormatException`. A non-numeric value like `abc` also blows up with a
raw stack trace instead of a user-facing error.
> * Impact: operators using the new `--restore-with-checkpoint` entrypoint
get an opaque `NumberFormatException` from deep inside `execute()` rather than
the friendly fail-fast message the PR advertises ("the submission fails fast").
The same untrimmed value would also be inconsistent with the trimmed value used
in the validation.
> * Fix: move the numeric validation into `buildCommand()` — trim the value
once, attempt `Long.parseLong(trimmed)` there, and throw
`IllegalArgumentException("--restore-with-checkpoint requires a numeric jobId,
got: ...")`. Store the trimmed value back so `ClientExecuteCommand` parses the
sanitized string. Consider applying the same guard to the pre-existing
`--restore` path for consistency.
> * Evidence:
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java:201
Long.parseLong(clientCommandArgs.getRestoreWithCheckpointJobId()),`
> * Suggested fix: This validation only catches a blank id, but
`ClientExecuteCommand` later runs `Long.parseLong()` on the raw, untrimmed
string. `--restore-with-checkpoint " 123 "` or `--restore-with-checkpoint abc`
passes here and then dies with a raw `NumberFormatException`. Please trim the
value and validate it is numeric here (e.g. try `Long.parseLong(trimmed)` and
throw an `IllegalArgumentException` with a clear message), so the CLI truly
fails fast with an actionable error.
> * Severity: Medium
>
> **Issue 4: checkpoint.retain-after-job-cancelled keeps serialized job
state on shared storage indefinitely with no retention policy, creating
data-at-rest exposure and unbounded disk growth**
>
> * Location:
`seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvCommonOptions.java:105`
> * Why it matters: - Root cause: the new
`checkpoint.retain-after-job-cancelled` option (job-level env override plus
cluster default) disables checkpoint cleanup on cancel, and the PR's own docs
state that checkpoints are NOT managed by `history-job-expire-minutes` and must
be cleaned up manually.
> * Impact: checkpoint files contain serialized source/sink state (CDC
offsets, split enumerator state, and in some connectors serialized config
fragments), so retaining them forever on HDFS/S3/local disk leaves sensitive
operational state at rest long after the job is gone; on busy clusters this
also grows storage unboundedly since cancelled jobs no longer self-clean.
Because the option is a job-level env key, any job submitter can force
retention without cluster-admin involvement, shifting the cleanup burden to
operators who may not know the option was set.
> * Fix: add a TTL/retention policy for retained checkpoints of cancelled
jobs (or fold them into `history-job-expire-minutes` handling), log a prominent
WARN with the checkpoint path when a job is cancelled with retention enabled,
and document a recommended cleanup procedure and the sensitivity of the
retained state.
> * Evidence:
`seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvCommonOptions.java:105
Options.key("checkpoint.retain-after-job-cancelled")`;
`docs/en/engines/zeta/state-storage-and-recovery.md:71 - By default, canceled
jobs still clean up existing checkpoint data. If you want to retain the`
> * Suggested fix: Retained checkpoints of cancelled jobs contain serialized
source/sink state (e.g. CDC offsets, enumerator internals) and — per the docs
in this PR — are not covered by `history-job-expire-minutes`, so enabling this
leaves sensitive state on shared storage indefinitely and grows disk usage
unboundedly. Please consider adding a retention/TTL policy for
retained-after-cancel checkpoints (or wiring them into the existing history
expiry), and emit a WARN with the checkpoint path when a job is cancelled with
retention enabled so operators know cleanup is now manual.
> * Severity: Medium
>
> **Issue 5: CLI restore-with-checkpoint job ID is parsed with raw
Long.parseLong, so non-numeric or whitespace-padded input crashes with an
unhelpful NumberFormatException; the blank-check trims but the parse does not**
>
> * Location:
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java:202`
> * Why it matters: - Root cause: `ClientCommandArgs.buildCommand()` only
rejects a blank value (`restoreWithCheckpointJobId.trim().isEmpty()`), while
`ClientExecuteCommand` later calls
`Long.parseLong(clientCommandArgs.getRestoreWithCheckpointJobId())` on the
untrimmed raw string. `Long.parseLong` rejects leading/trailing whitespace and
any non-numeric characters.
> * Impact: a user running `seatunnel.sh --restore-with-checkpoint " 123"`
or `--restore-with-checkpoint abc` passes the buildCommand validation but the
client blows up deep in `execute()` with a bare `NumberFormatException`,
instead of the documented "fail fast with a clear message" behavior. The same
pattern applies to the `Long.parseLong(clientCommandArgs.getCustomJobId())`
branch added on the same lines.
> * Fix: validate numeric format (e.g. `StringUtils.isNumeric(trimmed)` or a
try/parse with a descriptive `IllegalArgumentException` like
"--restore-with-checkpoint requires a numeric jobId, got: ...") inside
`buildCommand()`, and store/pass the trimmed value so the parse site always
receives a canonical numeric string.
> * Evidence:
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java:202
clientCommandArgs.getRestoreWithCheckpointJobId()),`;
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgs.java:159
if (restoreWithCheckpointJobId != null &&
restoreWithCheckpointJobId.trim().isEmpty()) {`
> * Suggested fix:
`Long.parseLong(clientCommandArgs.getRestoreWithCheckpointJobId())` will throw
a raw `NumberFormatException` for non-numeric input, and even for a value like
`" 123"` because the blank-check in `buildCommand()` trims while this parse
does not. Please validate/trim the jobId in `buildCommand()` (e.g. reject
non-numeric values with a clear `IllegalArgumentException`) so the CLI fails
fast with an actionable message instead of a stack trace. The
`Long.parseLong(clientCommandArgs.getCustomJobId())` branch just below has the
same issue.
> * Severity: Medium
>
> **Issue 6: --savepoint combined with --restore-with-checkpoint is silently
ignored; mutual-exclusion check only covers --restore**
>
> * Location:
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgs.java:155`
> * Why it matters: - Root cause: `buildCommand()` only guards `restoreJobId
!= null && restoreWithCheckpointJobId != null`. In
`ClientExecuteCommand.execute()` the new branch is an `else if` chained after
the existing savepoint/restore branches, so if a user passes both `--savepoint
<id>` and `--restore-with-checkpoint <id>`, the earlier branch wins and the
checkpoint-restore request is silently dropped.
> * Impact: an operator who mixes these flags gets no error and the job does
something different from what they asked for — a classic happy-path-only
validation gap. For a state-recovery feature, silently taking the wrong code
path can lead to jobs starting from the wrong state or triggering an unintended
savepoint.
> * Fix: extend the validation in `buildCommand()` to reject
`--restore-with-checkpoint` combined with `--savepoint` (and any other mutually
exclusive action flags such as `--cancel`/`--metrics` if they follow the same
branch chain), mirroring the existing `--restore` exclusivity check.
> * Evidence:
`seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java:194
} else if (null != clientCommandArgs.getRestoreWithCheckpointJobId()) {`
> * Suggested fix: The exclusivity check only covers `--restore` vs
`--restore-with-checkpoint`. Because the new branch in `ClientExecuteCommand`
is an `else if` after the existing branches, passing `--savepoint 100
--restore-with-checkpoint 200` will silently ignore the checkpoint-restore
request. Please also reject `--restore-with-checkpoint` when `--savepoint` (and
other exclusive action flags) is set, so we fail fast instead of taking an
unintended code path.
> * Severity: Medium
>
> **Issue 7: Retention config key is inconsistent between docs: job-level
env key is `checkpoint.retain-after-job-cancelled` but state-storage doc tells
users to set `retain-after-job-cancelled: true` "in the job env"**
>
> * Location: `docs/en/engines/zeta/state-storage-and-recovery.md:73`
> * Why it matters: - Root cause: the new env option is registered as
`checkpoint.retain-after-job-cancelled` in `EnvCommonOptions`, and the
cluster-level YAML key is `retain-after-job-cancelled` nested under
`checkpoint`. The English state-storage doc collapses both into a single key
name, telling users to set `retain-after-job-cancelled: true` either in the job
`env` or in `seatunnel.yaml`.
> * Impact: a user copying `retain-after-job-cancelled: true` into their job
`env` block will get an unrecognized option (or a silently ignored one), so
their canceled job's checkpoints get cleaned up and the checkpoint-restore
workflow fails — exactly the failure this feature is meant to prevent.
> * Fix: spell out both forms explicitly in the doc: job-level `env {
checkpoint.retain-after-job-cancelled = true }` versus cluster-level
`seatunnel.engine.checkpoint.retain-after-job-cancelled: true`, matching the
wording already used in the rest-api-job-lifecycle doc.
> * Evidence: `docs/en/engines/zeta/state-storage-and-recovery.md:73
`retain-after-job-cancelled: true`either in the job`env`or
in`seatunnel.yaml`.`;
`seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvCommonOptions.java:105
Options.key("checkpoint.retain-after-job-cancelled")`
> * Suggested fix: The key name here is ambiguous: in the job `env` the
option is `checkpoint.retain-after-job-cancelled` (per `EnvCommonOptions`),
while `retain-after-job-cancelled` is only valid nested under `checkpoint:` in
`seatunnel.yaml`. As written, a user pasting `retain-after-job-cancelled: true`
into their job env would have retention silently not applied and lose the
ability to restore. Please show both exact forms explicitly.
> * Severity: Low
>
> **Issue 8: CLI tests miss negative cases for non-numeric restore ids and
for the --savepoint + --restore-with-checkpoint combination**
>
> * Location:
`seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java:135`
> * Why it matters: - The new tests cover happy-path parsing, the
`--restore`/`--restore-with-checkpoint` conflict, and a blank id, which is good.
> * However, there is no test asserting behavior for a non-numeric id (e.g.
`--restore-with-checkpoint abc`) or a whitespace-padded id (`" 123 "`), both of
which currently blow up later with a raw `NumberFormatException` in
`ClientExecuteCommand`, nor for mixing `--savepoint` with
`--restore-with-checkpoint`, which is currently silently ignored.
> * Once the validation gaps are fixed in `buildCommand()`, please add tests
for these cases so the fail-fast contract ("the submission fails fast") is
locked in by the test suite.
> * Evidence:
`seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java:135
public void testRejectBlankRestoreWithCheckpointJobId() {`
> * Suggested fix: Nice coverage of the blank-id case. Could you also add
negative tests for a non-numeric id (`--restore-with-checkpoint abc`), a
whitespace-padded id (`" 123 "`), and the `--savepoint` +
`--restore-with-checkpoint` combination? Today those either throw a raw
`NumberFormatException` or are silently ignored downstream, and tests here will
pin down the intended fail-fast behavior once validation is added.
> * Severity: Low
>
> # Review conclusion
> ### Conclusion: can merge after the blocking items are fixed
> **1. Blocking items**
>
> * Issue 1: Possible public API change in
seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/protocol/codec/SeaTunnelGetJobCheckpointCodec.java
>
> **2. Suggested (non-blocking) follow-ups**
>
> * Issue 2: REST restore path lets any caller restore an arbitrary
historical job's checkpoint state via restoreSourceJobId with no
ownership/authorization check evident
> * Issue 3: Non-numeric or whitespace-padded --restore-with-checkpoint job
id passes buildCommand() validation but crashes later with a raw
NumberFormatException
> * Issue 4: checkpoint.retain-after-job-cancelled keeps serialized job
state on shared storage indefinitely with no retention policy, creating
data-at-rest exposure and unbounded disk growth
> * Issue 5: CLI restore-with-checkpoint job ID is parsed with raw
Long.parseLong, so non-numeric or whitespace-padded input crashes with an
unhelpful NumberFormatException; the blank-check trims but the parse does not
> * Issue 6: --savepoint combined with --restore-with-checkpoint is silently
ignored; mutual-exclusion check only covers --restore
> * Issue 7: Retention config key is inconsistent between docs: job-level
env key is `checkpoint.retain-after-job-cancelled` but state-storage doc tells
users to set `retain-after-job-cancelled: true` "in the job env"
> * Issue 8: CLI tests miss negative cases for non-numeric restore ids and
for the --savepoint + --restore-with-checkpoint combination
>
> Nice progress overall — once the blocking points above are addressed this
should be in good shape. Happy to discuss.
Thanks for the review. I went through each point again and addressed the
CLI/doc items in this update.
- Issue 1: I do not think this is a public API/SPI break.
SeaTunnelGetJobCheckpointCodec is an internal engine protocol codec, all
in-repo call sites were updated together, and codec tests were added for
coverage.
- Issue 2: Agreed, this is a valid hardening concern. The current
implementation does not add job-level authorization for restoreSourceJobId, so
this should be handled as a follow-up authz/audit improvement on the server
side.
- Issue 3 / 5 / 6 / 8: Agreed. Issues 3 and 5 are the same validation gap,
and I fixed them together with Issue 6 and 8 in one pass:
--restore-with-checkpoint is now trimmed and validated in buildCommand(),
--set-job-id gets the same numeric validation, --savepoint and
--restore-with-checkpoint are now mutually exclusive, and the missing negative
tests were added.
- Issue 4: Agreed on the risk. This is more of a retention/operations
concern than a functional bug in the restore path, and it should be addressed
with clearer cleanup guidance or a follow-up retention strategy.
- Issue 7: Agreed. The docs mixed the job-level env key with the
cluster-level YAML key, and I updated them to show both exact forms explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
