[FLINK-39891] Add option to cancel orphaned session jobs on submit#1135
[FLINK-39891] Add option to cancel orphaned session jobs on submit#1135lrsb wants to merge 1 commit into
Conversation
Dennis-Mircea
left a comment
There was a problem hiding this comment.
This PR looks like complementary to #1065 by addressing the same symptom but with different root cause. I'd say is worth linking both PRs + FLINK-38858 JIRA in the description so the scope is clear.
Two concerns on the mechanism:
-
Cancelling by
pipeline.nameis risky. Multiple jobs can share a name and the cluster doesn't know which "belongs" to this CR, so this can kill an unrelated job. A deterministic JobID (from the CR identity) would let Flink reject the resubmit instead, no listing/cancelling needed. It would also mean no new config property: fixing the root cause makes the protection automatic, whereas this is a correctness fix gated behind an opt-in flag that defaults off, so most users stay exposed. A correctness fix arguably shouldn't be opt-in. If a deterministic JobID won't work for the CR-recreate case, worth saying why. -
The description says
pipeline.namedefaults to the namespace-qualified name, but the code uses the baregetMetadata().getName(), so two same-named jobs in different namespaces on one cluster would cancel each other. Either fix the default or the description, and add a cross-namespace test.
What is the purpose of the change
This pull request guards against orphaned/duplicate session jobs. FLINK-38858
already prevents duplicates when the recorded
JobIDsurvives a failed submission(it reuses the
JobID, and Flink rejects a same-JobIDresubmit). A residualwindow remains where the operator generates a new
JobIDwhile a job from aprevious submission is still running on the session cluster - e.g. the submit
succeeded on the cluster but the
JobIDwas never durably persisted, or theFlinkSessionJobCR was deleted and recreated against the long-lived sessioncluster. The old job is then left orphaned, and two jobs run concurrently against
the same sources/sinks, breaking exactly-once semantics.
This change adds an opt-in safeguard: before a fresh submission, the operator
cancels any matching job already running on the session cluster.
Brief change log
kubernetes.operator.session-job.cancel-orphaned-on-submit(Boolean, default
false)pipeline.nameto the namespace-qualifiedFlinkSessionJobresourcename so a session job has a stable identity on the cluster
SessionJobReconciler.deploylists non-terminal jobs on the session clusterand, when the option is enabled and a new
JobIDis being generated (notreusing an existing one), cancels matching jobs and waits for them to reach a
terminal state before submitting
JobIDs are logged at WARNVerifying this change
This change added tests and can be verified as follows:
cluster and the option enabled, the orphan is cancelled before the new
submission; with the option disabled, behavior is unchanged
JobID-reuse tests continue to pass,confirming no behavior change when the option is off
Does this pull request potentially affect one of the following parts:
CustomResourceDescriptors: no(
SessionJobReconciler.deploy)Documentation