Repository navigation
Conversation
…stopping a task
Stopping a Flink task only killed the client process tree, and the cancel path
of FlinkStreamTask executed `${FLINK_HOME}/bin/flink cancel <id>` with
${FLINK_HOME} left unresolved, because ProcessBuilder doesn't expand shell
variables. So the Flink job could keep running after the task was stopped: the
batch task never issued `flink cancel` at all, and the stream task always failed
with `Cannot run program "${FLINK_HOME}/bin/flink"`.
- resolve ${FLINK_HOME} for the cancel and savepoint commands (task params
first, then the FLINK_HOME environment variable)
- move the cancel logic from FlinkStreamTask into FlinkTask, so batch and stream
tasks share it: cancel by the JobID printed by `flink run` and fall back to the
previous behaviour (kill the process tree and cancel the YARN/K8s application)
when there is no JobID, when the task was submitted to a resource manager, or
when the CLI fails
- wait for the cancel command to finish instead of fire-and-forget, so a failure
can be detected and fall back
- add unit tests for the command resolution and the cancel/fallback logic
Verified by driving the real classes in a Linux container: a batch task
(deployment cluster, detached submit) and a stream task both execute
`flink cancel <jobId>` now, where previously the batch task never issued it and
the stream task always threw.
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Use the task environment for cancellation.
FlinkArgsUtils.buildFlinkCommand() only reads prepared parameters and System.getenv(), while submission also loads shell.env_source_list and TaskExecutionContext.environmentConfig. If FLINK_HOME is defined only in the selected task environment, submission succeeds but cancellation attempts the literal "${FLINK_HOME}/bin/flink" and fails. Please reuse the submission environment and tenant context.
Preserve cancellation failures and keep JobIDs separate from application IDs.
When the Flink CLI fails, FlinkTask.cancelApplication() falls back to killing the client and performing resource-manager cleanup. This cannot cancel a standalone/session Flink job. Moreover, cancelFlinkJob() has already overwritten context.appIds with the Flink JobID, so the fallback passes it to the YARN application manager. Please keep these identifiers separate and propagate the failure when the remote job cannot be cancelled.
Drain or redirect the CLI output.
FlinkArgsUtils.executeCommand() waits for completion without consuming stdout or stderr. If either pipe fills, the CLI blocks until the 30-second timeout and is forcibly terminated. Please drain or redirect both streams and retain the output for failure diagnostics.
…path
- run the cancel / savepoint command in the same environment as the task.
The command is now executed by a shell interceptor which sources
shell.env_source_list, applies the task's custom environment and runs as the
task's tenant, so ${FLINK_HOME} is resolved even when it is only defined in
the selected task environment. The Java-side placeholder resolution is dropped.
- keep the Flink JobID and the YARN/K8s application id separate. The JobID is no
longer written into taskExecutionContext.appIds, so the fallback cannot pass it
to the YARN application manager, and a failed cancel is now propagated instead
of being hidden behind the fallback that cannot cancel a standalone/session job.
- consume the output of the cancel command so it cannot block on a full pipe, and
keep the output for failure diagnostics.
- add a test that executes the cancel command with FLINK_HOME defined only in the
task environment and writes more output than a pipe buffer holds.
@SbloodyS All three points are fixed in 691373a:
|
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Propagate savepoint command failures.
In FlinkStreamTask.java:77, savePoint() ignores the boolean returned by FlinkArgsUtils.executeCommand(). That method returns false on startup errors, non-zero exit codes, timeouts, and interruption. Consequently, savePoint() returns normally and StreamingTaskInstanceOperatorImpl.triggerSavepoint() reports success even when the savepoint command failed.
Please throw TaskException when executeCommand() returns false, and cover the failure response in the savepoint tests.
…ream task savePoint() ignored the result of the savepoint command, so StreamingTaskInstanceOperatorImpl reported success even when the command failed to start, exited with a non-zero code, timed out or was interrupted. It also returned normally when no application id could be collected, without triggering any savepoint at all. - throw TaskException when the savepoint command fails - throw TaskException when no application id can be found, instead of reporting a successful savepoint that never ran - cover the savepoint command, the failed command and the missing application id in FlinkStreamTaskTest
|
@SbloodyS Fixed in e5abd15: savePoint() now throws TaskException when the savepoint command returns false, and also when the application id is missing (it used to return silently and report success). Failure cases are covered in FlinkStreamTaskTest. Full run: flink 16 tests and flink-stream 12 tests pass on Linux, spotless passes. |
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Correct the JobID used by the new savepoint success test.
FlinkStreamTaskTest.testSavePointCommand() expects:
flink savepoint application_1700000000000_0001
However, flink savepoint requires a Flink JobID; the YARN application ID is a separate parameter. Mocking executeFlinkCommand() to return true makes this invalid command appear successful.
This identifier problem predates the PR, but the new test now locks in the incorrect behavior. Please use a real Flink JobID in the success test and obtain it through getFlinkJobIds(), keeping any YARN application ID separate.
`flink savepoint` takes the Flink JobID, not the YARN/K8s application id. FlinkStreamTask.savePoint() used the application id returned by getApplicationIds(), which cannot work for standalone/session deployments, and the new savepoint test locked in a command that could never succeed. - buildSavePointCommandLine() now takes the JobID explicitly - savePoint() resolves the JobID through getFlinkJobIds() and no longer reads or writes taskExecutionContext.appIds, so the two identifiers stay separate - tests use a real JobID, cover the case where only a YARN application id is present, and assert the application id is never passed to `flink savepoint`
…nd test testExecuteCommandReuseTaskEnvironment does not depend on Linux specific tooling, it only runs a bash script through the shell interceptor. The whole FlinkArgsUtilsTest passes on macOS as well (9/9), so the condition can be dropped and the test also runs for developers on macOS.
|
@SbloodyS Fixed in 1f35037: savePoint() now uses the Flink JobID from getFlinkJobIds() and passes it to buildSavePointCommandLine(jobId); the YARN application id is no longer used or written back. The success test uses a real JobID (with the application id present in the log to prove it is ignored), and the missing-id test asserts no CLI call and a TaskException. flink 16 + flink-stream 12 tests pass on Linux, spotless passes. |
…he command shell CodeQL reported a new use of the deprecated TaskExecutionContext.getTaskAppId() in the cancel / savepoint command execution. The field is deprecated in favour of the task instance id, and the shell name is only a file name inside the task working directory, which is already scoped to the task instance, so the task instance id is equivalent here. No behaviour change: flink 16 + flink-stream 12 tests still pass, spotless passes.
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Parse JobIDs emitted by the SQL Client
FlinkTask.java:191–195 reuses findAppId(), whose regex only matches "JobID ". Flink SQL Client prints "Job ID: ", so SQL submissions are ignored. Cancellation then falls back to killing the client, and savePoint() reports a missing JobID.
Please support both output formats and add regression tests using actual SQL-client output.
Reference: https://github.com/apache/flink/blob/release-1.16/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java#L587-L592
Preserve cluster targeting when triggering savepoints
FlinkStreamTask.java:75–79 constructs only "flink savepoint ". For a YARN application without a matching local session-properties file, the CLI also needs the application ID as a targeting option (-yid); the JobID alone does not identify the cluster.
Please keep the identifiers separate while passing the appropriate cluster-targeting options. The current test includes a YARN application ID but mocks executeFlinkCommand() to return true, so it cannot detect this failure.
Reference: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/state/savepoints/#trigger-a-savepoint-with-yarn
…avepoint
- match both the `flink run` JobID format ("JobID <id>") and the Flink SQL
Client format ("Job ID: <id>"), so SQL submissions can be cancelled and
savepointed instead of falling back to killing the client
- read the JobID from the regex capture group instead of a hardcoded
substring(6)
- pass the YARN application id as the `-yid` cluster targeting option when
triggering a savepoint, keeping it separate from the Flink JobID
- cover both JobID formats and both savepoint targeting cases in the tests
flink (18) + flink-stream (14) tests pass, spotless passes.
…on' into fix/flink-16789-cancel-application
|
@SbloodyS Thanks for the review. Both points are addressed in 1. Parse JobIDs emitted by the SQL Client
Regression tests use the actual SQL-client line:
2. Preserve cluster targeting when triggering savepoints
The tests now assert the full argument list instead of relying on the mocked
Verification
All the earlier comments (task environment reuse, cancellation failure propagation, CLI output draining, savepoint failure propagation, JobID vs application id separation) are addressed in the same branch. |
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Preserve the submission's cluster connection options
Submission appends flinkParameters.getOthers(), allowing a standalone job to target a different JobManager with, for example, "-m remote-jm:8081". However, buildCancelCommandLine() only emits "flink cancel ", and buildSavePointCommandLine() only preserves the YARN application ID.
Consequently, a job submitted to that remote JobManager is cancelled or checkpointed against the default cluster instead. Cancellation fails and the remote job remains running. Reusing the task environment does not preserve command-line connection options.
Please carry the applicable cluster connection options into cancel/savepoint commands, without copying submission-only arguments. Add coverage where the submission endpoint differs from the default Flink configuration.
…link cancel and savepoint commands
The submission appends the extra options of the task ("-m <jobmanager>",
"-t <target>", "-D<key>=<value>"), but the cancel and savepoint commands only
emitted the job id. A job submitted to a remote JobManager was therefore
cancelled or savepointed against the default cluster of the worker.
- extract the cluster connection options from the extra options of the
submission, dropping submission-only arguments such as the parallelism, the
slots, the memory, the application name and the main class
- repeat them on the cancel and savepoint commands
- cover a submission endpoint which differs from the default Flink
configuration
flink (24) + flink-stream (15) tests pass, spotless passes.
…d fails and rename the JobID helpers cancelApplication() propagated the failure of the cancel command before stopping the client process, so the client kept running and the worker could not bring the task instance to a final state. The client is now stopped even when the remote job could not be cancelled, and the failure is still propagated afterwards. - always stop the client through the new cancelClientByProcess() seam, then throw the recorded failure instead of skipping the cleanup - rename findAppId / FLINK_APPLICATION_REGEX to findFlinkJobId / FLINK_JOB_ID_REGEX, so the JobID is no longer named after the YARN/K8s application id - only repeat the -D options which identify the cluster, and tokenize the extra options with quote awareness, so a quoted value is not split - describe the real behaviour in the cancelApplication javadoc flink (28) + flink-stream (15) + task-api (114) tests pass, spotless passes.
… tighten the JobID parsing - `flink savepoint` no longer shares the 30s cancel timeout: taking a savepoint of a large stateful job can take much longer, so it uses FLINK_SAVEPOINT_TIMEOUT_SECONDS and its own execution seam - require the 32 hexadecimal characters of a Flink JobID, so an unrelated line which contains "Job ID" cannot be collected as a JobID - correct the cancelFlinkJob javadoc: only YARN submissions are handed to the resource manager, a Kubernetes job is cancelled through the CLI - cover the command timeout with a test, which needs the timeout to be injectable task-api (114) + flink (29) + flink-stream (15) tests pass, spotless passes.
|
@SbloodyS Addressed in e6b0ca6 (plus follow-ups e109ceb, ff0809a); branch now merged with latest dev (3f3d010). |
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Preserve quoted segments inside option tokens
FlinkArgsUtils.java:187–193: the tokenizer handles quotes only when they begin a token. For example, -Dkubernetes.config.file="/opt/flink configs/kube.yaml" is split into -Dkubernetes.config.file="/opt/flink and configs/kube.yaml".
The filter retains only the first fragment, so the generated cancel/savepoint script contains an unmatched quote and fails before invoking Flink. Submission preserves the original option and can succeed.
Please support quotes and escapes within tokens, and add coverage for attached and separated -D values containing spaces.
Terminate the command's process tree on timeout
FlinkArgsUtils.java:329–335: the process returned by the shell interceptor is the bootstrap shell or sudo process. Calling destroyForcibly() only on that process does not terminate its descendants, including the Flink CLI JVM.
Consequently, cancellation/savepoint can report a timeout while the CLI continues running. Descendants can also retain stdout and keep the output-reader thread alive.
Please terminate and await the command's process tree on timeout and interruption. The timeout regression should verify descendant termination, rather than only checking the method's return time.
…ut and keep quoted option segments - terminate the process tree of the flink command on timeout and interruption instead of only the bootstrap process, and wait for it to be gone: the CLI runs as a descendant of the shell, so killing the shell left the CLI running and holding the output pipe - add ProcessUtils.getProcessId(Process) and ProcessUtils.kill(processId, tenantCode), and reuse them from AbstractCommandExecutor instead of duplicating the pid reflection - tokenize the extra options with quote awareness anywhere inside a token, so -Dkubernetes.config.file="/opt/flink configs/kube.yaml" stays a single token - cover the process tree termination (delegation and real descendant) and the quoted values task-api (114) + flink (33) + flink-stream (15) tests pass, spotless passes.
…mand process tree and recognize fully quoted options - fall back to `ps -eo pid,ppid` when pstree is missing or returns nothing, so the process tree of a command is still terminated on a host without pstree; previously only the bootstrap shell was destroyed there and the descendants survived - drop the assumption from the descendant termination test: it now runs everywhere and asserts that the background descendant is gone - recognize an option or a value which is entirely quoted, for example `-D "key=a b"` and `"-Dkey=a b"`, instead of silently dropping it task-api (114) + flink (34) + flink-stream (15) tests pass, spotless passes.
|
@SbloodyS Both fixed in d70ba26.
|
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
- Stop the submission client even after successful remote cancellation
FlinkTask.java:144–145
Returning immediately when cancelFlinkJob() succeeds leaves the original client process running. For a multi-statement SQL script, cancellation only covers the JobIDs collected from the log at that moment; the SQL Client can continue executing later statements and submit another job after the stop request. Ensure that stopping the task also terminates the submission client, and cover successful remote cancellation while the client still has pending statements.
- Attempt cancellation of every collected JobID
FlinkTask.java:203–209
The loop throws on the first failed cancellation, leaving all subsequent jobs untouched. For example, a SQL script can have a completed batch job followed by a running streaming job. Cancelling the completed JobID can fail, so the running job is never cancelled; killing the client afterward does not stop that remote job. Attempt every cancellation and aggregate failures afterward, while distinguishing already-terminal jobs from actual cancellation failures. Add coverage with two JobIDs where the first cancellation fails.
- Preserve REST TLS settings in control commands
FlinkArgsUtils.java:84–96
The connection-option whitelist drops settings such as security.ssl.rest.enabled and security.ssl.rest.truststore. A submission using "-m remote-jm:8081 -Dsecurity.ssl.rest.enabled=true" can connect successfully, but cancellation/savepoint retains the address while reverting to the default non-TLS connection. Preserve the applicable connection security settings alongside the endpoint, and add regression coverage for TLS options supplied through others.
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
…ped and preserve the REST TLS options - stop the submission client even when the remote job was cancelled: a multi statement SQL script can keep submitting jobs after the cancelled statement - attempt every collected JobID and aggregate the failures afterwards, and do not report a job which is already in a terminal state as a cancellation failure (the flink CLI reports "already reached another terminal state") - keep the security.ssl.rest. configuration on the cancel / savepoint command, so a submission over a TLS REST endpoint does not fall back to a plain connection - keep the output of a command so that a failure can be classified, and stop logging the command arguments which may carry sensitive configuration values task-api (114) + flink (36) + flink-stream (15) tests pass, spotless passes.
|
@SbloodyS All three addressed in 12c73ab.
|
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Stop the submission client before taking the final JobID snapshot
[Source-based finding] FlinkTask.java:152–160
cancelFlinkJob() snapshots JobIDs while the SQL Client is still running, then waits for each cancellation before stopping the client. During that interval, the client can submit another job that is absent from the snapshot. Killing the client afterward leaves that remote job running.
Please prevent further submissions before collecting the final JobIDs, account for buffered submission output, and add coverage for a job submitted during cancellation.
Preserve cluster settings configured through SQL SET statements
[Source-based finding] FlinkTask.java:267–272
Connection options are extracted exclusively from others, although SQL submission also executes initScript and rawScript. For example, a standalone SQL task can select its remote endpoint with:
SET 'rest.address' = 'remote-jm';
SET 'rest.port' = '8081';Cancel/savepoint then uses the worker's default endpoint and fails to control the submitted job.
Please retain the effective connection configuration for each submitted job and add coverage for settings supplied through SQL initialization.
…JobIDs and keep the SQL connection settings - stop the client before the JobIDs are collected: while it is running it can submit further jobs which an earlier snapshot would miss, and wait (bounded) for the buffered submission output to reach the task log before collecting them - collect the cluster connection settings from the SET statements of the SQL scripts as well, and merge them with the extra options, the effective value of a key wins - cover a job submitted while the client is stopped, and the settings supplied through the SQL scripts task-api (118) + flink (40) + flink-stream (15) tests pass, spotless passes.
…ling the log and match only SQL SET statements - expose AbstractCommandExecutor.awaitOutputCollected(timeout) and use it from AbstractYarnTask, so the cancel path waits for the output which the worker collects asynchronously instead of polling the log file length, which could return before the tail of the output was written - only match a SET statement at the beginning of a statement, so a SET which appears inside a string literal or a comment is not collected as a connection option, and describe the four capture groups correctly - cover the literal, comment and double quoted forms task-api (118) + flink (41) + flink-stream (15) tests pass, spotless passes.
…he SQL SET limitations - extract AbstractCommandExecutor.awaitCollected and cover the collected, missing, failed and timed out collection, so the wait introduced for the cancel path is no longer untested (task-api tests 118 -> 122) - add a FlinkTask seam for the wait and assert with Mockito.inOrder that the cancel path stops the client, waits for the collected output and only then cancels - state the two known limitations of the SQL SET extraction in the javadoc: a multi line string literal whose line begins with SET is still matched, and a script which changes the connection between two submitted jobs is represented by the value of the last statement task-api (122) + flink (41) + flink-stream (15) tests pass, spotless passes.
|
@SbloodyS Both findings are fixed in 1. Stop the submission client before taking the final JobID snapshot
2. Preserve cluster settings configured through SQL SET statements
Covered by Known limitations, documented in the javadoc: a multi line string literal whose line begins with Verification
|
This branch has not been deployed
3 participants
Was this PR generated or assisted by AI?
YES. Prepared with AI assistance, using DSH (DeepSeek Harness) as the coding agent; the change was built, formatted and tested locally (
mvn clean test+spotless:checkon both Flink task modules, 24 tests) before submission.Purpose of the pull request
Fix #16789. Stopping a Flink task killed only the client process tree, so the Flink job could keep running afterwards:
FlinkTasknever issuedflink cancel;FlinkStreamTaskdid, but with an unresolved${FLINK_HOME}— the cancel command is executed byProcessBuilder, which does not expand shell variables — so it always failed withCannot run program "${FLINK_HOME}/bin/flink".Brief change log
${FLINK_HOME}for the cancel / savepoint commands (task params first, then theFLINK_HOMEenvironment variable)FlinkStreamTaskintoFlinkTask: cancel by the JobID printed byflink run, and fall back to the previous behaviour (kill the process tree, cancel the YARN/K8s application) when the task was submitted to a resource manager, when there is no JobID, or when the CLI failsflink cancelbefore the process killVerify this pull request
This change added tests and can be verified as follows:
FlinkArgsUtilsTest:${FLINK_HOME}is resolved in the cancel / savepoint commandFlinkTaskTest: cancel via the CLI using the JobID from the task log; fallback when the CLI fails; fallback when no JobID is found; no CLI call for YARN/K8s submissionsdeployMode=cluster, detached submit) and a stream task both executeflink cancel <jobId>now, where before the batch task never issued it and the stream task always threwPull Request Notice
Pull Request Notice