[Flyer4109]

Overview

Add spec.jobManager.deploymentType (Job | Deployment). With Deployment, the Application-mode JobManager runs as a Kubernetes Deployment instead of a Job, so Kubernetes recreates a lost JobManager pod and Flink HA recovers the job. The default stays Job, which behaves exactly as before.

Why

Today the JobManager is a Job with restartPolicy: Never, so a lost JobManager pod comes back only as a side effect of restartPolicy: FromSavepointOnFailure. After a terminal failure, that policy restarts from an old savepoint and loses data that was already acknowledged. We want a real JobManager recovery mechanism so we can stop relying on it for streaming. ES-1028

Description

• API: JobManagerDeploymentType (Job default when unset, read through GetDeploymentType()). Deployment requires Application mode and HA. The field is immutable, enforced by the webhook plus a CEL rule on jobManager that compares effective values.
• restartPolicy is ignored with Deployment: ShouldRestart returns false, so a failed job is never restored from a savepoint. This is accepted rather than rejected because admission still sets FromSavepointOnFailure during the rollout. ShouldRestart and IsTerminated now take the FlinkClusterSpec.
• Converter:
  • The JobManager Deployment has 1 replica, strategy Recreate, pod restartPolicy: Always, and the same name, labels and pod spec as the JobManager Job.
  • The operator sets execution.shutdown-on-application-finish: false.
• Observer: reads the Deployment and selects the newest pod that is not being deleted. Job mode keeps its current pod selection.
• Updater:
  • Terminal states come only from the Flink API, and the recorded state is kept while the API is unavailable.
  • A recently started JobManager gets a 5 min grace period to recover the job from HA before the job is marked Lost.
  • A Lost job is tracked again once the Flink API reports it, or cleaned up with AfterJobFails after 10 min.
  • A crash-looping JobManager (CrashLoopBackOff after 3 short runs) marks a deploying job DeployFailed and a running job Failed, with a failure reason.
  • If the Deployment of an active job is missing, the job goes back to Pending so the Deployment is created again.
• Terminated jobs: the Deployment is scaled to 0. The object stays without a pod, like the finished JobManager Job, and is deleted only by DeleteCluster.
• Reconciler: shared create/delete helpers for the job workload cover create, update (delete the old revision after stop-with-savepoint, then create the next from the savepoint, then Stopped → Deploying), scale-down and deletion.
• Volcano: schedules the Deployment's pod.

Merge order: this PR → operator release v0.7.0 → flink-admission PR → flink-admission release v0.4.13 → cluster-config C → D → E (see ES-1028).

Test Plan

• make test passes (unit tests cover both modes)
• make verify-helm-crd-schema passes
• CI green

[grzegorz8]

Testing results

Test scenarios

#	Scenario	Description	Steps	Deployment - Flink 2.2.1	Deployment - Flink 2.3.0	Job - Flink 2.2.1	Job - Flink 2.3.0
1	Regular update	Restore a savepoint after a regular update.	1. Start the job; change a non-scaling flinkProperties value. 2. Verify the final savepoint succeeds. 3. Verify the replacement restores that savepoint with the same job ID, reaches Running, and revisions converge. For Deployment, verify all old JM pods disappear before replacement; Job does not require that ordering. 4. Repeat the update once.	Pass	Pass	Pass	Pass
2	Cancel with savepoint	Stop intentionally and apply cancellation cleanup.	1. Start the job and request job-cancel. 2. Verify final savepoint and control succeed; job becomes Cancelled. 3. Verify cancellation cleanup and no resubmission.	Pass	Pass	Pass	Pass
3	Non-stopping savepoint	Snapshot without interrupting execution.	1. Set annotation flinkclusters.flinkoperator.k8s.io/user-control=savepoint on the FlinkCluster. 2. Verify savepoint/control succeed and the snapshot is not final. 3. Verify the job stays Running, processing continues, and JM workload/pod UIDs remain unchanged.	Pass	Pass	Pass	Pass
4	Delete running FlinkCluster	Finalize deletion after JM termination.	1. Delete the running CR without cancelling first. 2. Verify the finalizer remains while JM pods terminate and no replacement appears. 3. Verify the CR and owned resources, including discovered HA ConfigMaps, are eventually removed; no final savepoint is expected.	Pass	Pass	Pass	Pass
5a	Exhaust Flink retries after checkpoint	Verify a completed checkpoint before injecting task failure.	1. Start capture; set the failure control file to RUN and verify it. 2. Submit; verify at least one completed checkpoint from this run. 3. Set FAIL and keep it through retry exhaustion. 4. Verify terminal Failed, cleanup to Stopped, and no resubmission for 2 minutes.	Pass	Pass	Pass	Pass
5b	Exhaust Flink retries from startup	Verify failure and retries when the control file is FAIL before submission.	1. Start capture; set the failure control file to FAIL before submission and verify it. 2. Submit once; no RUN phase or checkpoint prerequisite. 3. Verify ControlledFailureException and exhaustion of configured retries. 4. Verify terminal Failed, cleanup to Stopped, and no resubmission for 2 minutes.	Pass	Pass	Pass	Pass
7	Delete JobManager pod	Check the outcome of losing the JM pod.	1. Delete the JM pod; for Deployment, force-delete it to skip preStop and verify the old process terminates. 2. Deployment: verify a replacement pod with a different UID, and recovery of the same job ID and checkpoint state despite temporary REST unavailability. 3. Job: under the Never restart policy, verify terminal Failed/Lost, failure cleanup, and no Deployment-style recovery or resubmission.	Pass	Pass	Pass*	Pass*
8	Delete JobManager Deployment	Recover through operator workload recreation.	1. Delete the running JM Deployment. 2. Verify the operator recreates it. 3. Verify the same job ID and checkpoint state recover without premature failure cleanup.	Pass	Pass	—	—
9	Exception in main()	Detect application submission failure.	1. Submit an application whose main() throws. 2. Verify a useful failure reason: Deployment via REST or crash-loop detection; Job via Kubernetes Job failure. 3. Verify failure cleanup and no indefinite Deploying. 4. Observe for 2 minutes; verify no resubmission.	Pass	Fail	Pass	Pass
11	Invalid submission artifact	Handle an invalid class or JAR.	1. Submit with an invalid main class or missing JAR. 2. Verify visible failure and failure cleanup. 3. Verify no unmanaged restart loop. The recorded results below cover the invalid entry class case.	Pass	Pass	Pass	Pass
12	Finite application succeeds	Accept genuine terminal success.	1. Run a finite application with afterJobSucceeds: DeleteCluster. 2. Verify Succeeded, success cleanup, and no resubmission.	Pass	Pass	Pass	Pass

Scenario 7: JM pod is expected to be restored for jobManager.deploymentType: Deployment, and for jobManager.deploymentType: Job with job.restartPolicy: FromSavepointOnFailure.

Issues found

deploymentType: Deployment — job stops during update

Test scenario: Update any of flinkProperties so a regular job update is triggered.

Expected behavior: The old job stops with savepoint, then a new job restores from the savepoint.

Actual behavior: The cluster reached Stopped state because the operator accepted the original execution’s FINISHED result while the replacement was still Deploying. It marked the job Succeeded and applied afterJobSucceeds: DeleteCluster.

Reason: jobManager.deploymentType: Deployment requires execution.shutdown-on-application-finish: false. As a result, the old JobManager does not shut down on its own after the stop-with-savepoint operation completes. Once the operator detects that the stop-with-savepoint has finished, it kills the old JobManager and schedules a new JobManager deployment.

However, each JobManager pod has lifecycle.preStop: sleep 30, which delays shutdown. Consequently, for a short period both the old and new JobManagers are running, and the old one remains the leader. The operator continues to query the old JobManager via the REST API and, when checking the job status, reads the status of the old, finished job. It then concludes that the job has reached a terminal state and cleans up the cluster.

Solution: Block replacement Deployment creation until all old JobManager pods, including terminating pods, disappear. Requeue while pods remain and propagate pod-list errors without creating the replacement.

Consequences: Job updates take longer, with the additional delay roughly equal to the preStop sleep duration. We can lower this duration to some extent, but it must remain long enough for the JobManager to complete its cleanup, especially of HA metadata.

---

Flink 2.3.0, deploymentType: Deployment — startup failure remains undetected

Test scenario: Submit an application whose main() throws before submitting a job graph, for example WindowJoin with --rate not-a-number.

Expected behavior: The operator detects the submission failure, records its cause, and applies afterJobFails: DeleteCluster, bringing the cluster to Stopped.

Actual behavior: Flink recorded the application as FAILED, but the JobManager remained running and the operator left the job in Deploying. Failure cleanup was never triggered.

Reason: Flink 2.3.0 tracks application execution separately from individual jobs. When main() throws before submitting a job graph, the application is marked FAILED, but the jobs endpoint returns an empty list. The operator queries job-level status, so it sees no failed job and leaves the submission in Deploying.

With execution.shutdown-on-application-finish: false, the JobManager stayed alive after the application failed. Consequently, neither job-level failure detection nor pod crash-loop detection exposed the failure to the operator.

Proposed solution:

Short-term solution: Set execution.submit-failed-job-on-application-error: true. However, the property is deprecated now.

If a failed job should be submitted (in the application mode) when there is an error in the application driver before an actual job submission. This is intended for providing a clean way of reporting failures back to the user and is especially useful in combination with 'execution.shutdown-on-application-finish'. This option only works when the single job submission is enforced ('high-availability.type' is enabled). Please note that this is an experimental option and may be changed in the future.

Long-term solution: Observe Flink 2.3.0 application-level status in addition to job status. When an application fails before submitting any jobs, record a terminal submission failure and its cause, then apply the configured failure cleanup policy.