Revert jobsubmitterfailed status and keep exitCode field (#427)

* Fix job state based on jobSubmitter exitCode

* Removed new JobStatus to keep things simple

Author: live-wire

apis/flinkcluster/v1beta1/flinkcluster_types.go

@@ -52,18 +52,17 @@ type JobMode string
 
 // JobState defines states for a Flink job deployment.
 const (
-	JobStatePending         = "Pending"
-	JobStateUpdating        = "Updating"
-	JobStateRestarting      = "Restarting"
-	JobStateDeploying       = "Deploying"
-	JobStateDeployFailed    = "DeployFailed"
-	JobStateSubmitterFailed = "SubmitterFailed"
-	JobStateRunning         = "Running"
-	JobStateSucceeded       = "Succeeded"
-	JobStateCancelled       = "Cancelled"
-	JobStateFailed          = "Failed"
-	JobStateLost            = "Lost"
-	JobStateUnknown         = "Unknown"
+	JobStatePending      = "Pending"
+	JobStateUpdating     = "Updating"
+	JobStateRestarting   = "Restarting"
+	JobStateDeploying    = "Deploying"
+	JobStateDeployFailed = "DeployFailed"
+	JobStateRunning      = "Running"
+	JobStateSucceeded    = "Succeeded"
+	JobStateCancelled    = "Cancelled"
+	JobStateFailed       = "Failed"
+	JobStateLost         = "Lost"
+	JobStateUnknown      = "Unknown"
 )
 
 // AccessScope defines the access scope of JobManager service.

apis/flinkcluster/v1beta1/flinkcluster_types_util.go

@@ -22,8 +22,7 @@ func (j *JobStatus) IsFailed() bool {
 	return j != nil &&
 		(j.State == JobStateFailed ||
 			j.State == JobStateLost ||
-			j.State == JobStateDeployFailed ||
-			j.State == JobStateSubmitterFailed)
+			j.State == JobStateDeployFailed)
 }
 
 func (j *JobStatus) IsStopped() bool {

controllers/flinkcluster/flinkcluster_converter.go

@@ -1065,7 +1065,7 @@ func shouldCleanup(cluster *v1beta1.FlinkCluster, component string) bool {
 	switch jobStatus.State {
 	case v1beta1.JobStateSucceeded:
 		action = cluster.Spec.Job.CleanupPolicy.AfterJobSucceeds
-	case v1beta1.JobStateFailed, v1beta1.JobStateLost, v1beta1.JobStateDeployFailed, v1beta1.JobStateSubmitterFailed:
+	case v1beta1.JobStateFailed, v1beta1.JobStateLost, v1beta1.JobStateDeployFailed:
 		action = cluster.Spec.Job.CleanupPolicy.AfterJobFails
 	case v1beta1.JobStateCancelled:
 		action = cluster.Spec.Job.CleanupPolicy.AfterJobCancelled

controllers/flinkcluster/flinkcluster_updater.go

@@ -620,8 +620,6 @@ func (updater *ClusterStatusUpdater) deriveJobStatus() *v1beta1.JobStatus {
 	switch {
 	case oldJob == nil:
 		newJobState = v1beta1.JobStatePending
-	case oldJob.SubmitterExitCode != newJob.SubmitterExitCode && isNonZeroExitCode(newJob.SubmitterExitCode):
-		newJobState = v1beta1.JobStateSubmitterFailed
 	case shouldUpdateJob(&observed):
 		newJobState = v1beta1.JobStateUpdating
 	case oldJob.ShouldRestart(jobSpec):