[FLINK-35776] Simplify job status handling

Author: gyfora

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/JobStatusUtils.java

@@ -38,16 +38,18 @@ public class JobStatusUtils {
     public static List<JobStatusMessage> toJobStatusMessage(
             MultipleJobsDetails multipleJobsDetails) {
         return multipleJobsDetails.getJobs().stream()
-                .map(
-                        details ->
-                                new JobStatusMessage(
-                                        details.getJobId(),
-                                        details.getJobName(),
-                                        getEffectiveStatus(details),
-                                        details.getStartTime()))
+                .map(JobStatusUtils::toJobStatusMessage)
                 .collect(Collectors.toList());
     }
 
+    public static JobStatusMessage toJobStatusMessage(JobDetails details) {
+        return new JobStatusMessage(
+                details.getJobId(),
+                details.getJobName(),
+                getEffectiveStatus(details),
+                details.getStartTime());
+    }
+
     @VisibleForTesting
     static JobStatus getEffectiveStatus(JobDetails details) {
         int numRunning = details.getTasksPerState()[ExecutionState.RUNNING.ordinal()];

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/exception/MissingSessionJobException.java

@@ -17,30 +17,29 @@
 
 package org.apache.flink.kubernetes.operator.observer;
 
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
 import org.apache.flink.kubernetes.operator.api.spec.JobState;
 import org.apache.flink.kubernetes.operator.api.status.JobStatus;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
 import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
 import org.apache.flink.kubernetes.operator.utils.EventRecorder;
 import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.rest.NotFoundException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.TimeoutException;
 
 import static org.apache.flink.kubernetes.operator.utils.FlinkResourceExceptionUtils.updateFlinkResourceException;
 
 /** An observer to observe the job status. */
-public abstract class JobStatusObserver<R extends AbstractFlinkResource<?, ?>> {
+public class JobStatusObserver<R extends AbstractFlinkResource<?, ?>> {
 
     private static final Logger LOG = LoggerFactory.getLogger(JobStatusObserver.class);
 
-    public static final String MISSING_SESSION_JOB_ERR = "Missing Session Job";
+    public static final String JOB_NOT_FOUND_ERR = "Job Not Found";
 
     protected final EventRecorder eventRecorder;
 
@@ -68,58 +67,51 @@ public boolean observe(FlinkResourceContext<R> ctx) {
         LOG.debug("Observing job status");
         var previousJobStatus = jobStatus.getState();
 
-        List<JobStatusMessage> clusterJobStatuses;
         try {
-            // Query job list from the cluster
-            clusterJobStatuses =
-                    new ArrayList<>(ctx.getFlinkService().listJobs(ctx.getObserveConfig()));
+            var newJobStatusOpt =
+                    ctx.getFlinkService()
+                            .getJobStatus(
+                                    ctx.getObserveConfig(),
+                                    JobID.fromHexString(jobStatus.getJobId()));
+
+            if (newJobStatusOpt.isPresent()) {
+                updateJobStatus(ctx, newJobStatusOpt.get());
+                ReconciliationUtils.checkAndUpdateStableSpec(resource.getStatus());
+                return true;
+            } else {
+                onTargetJobNotFound(ctx);
+            }
         } catch (Exception e) {
             // Error while accessing the rest api, will try again later...
-            LOG.warn("Exception while listing jobs", e);
+            LOG.warn("Exception while getting job status", e);
             ifRunningMoveToReconciling(jobStatus, previousJobStatus);
             if (e instanceof TimeoutException) {
                 onTimeout(ctx);
             }
-            return false;
-        }
-
-        if (!clusterJobStatuses.isEmpty()) {
-            // There are jobs on the cluster, we filter the ones for this resource
-            Optional<JobStatusMessage> targetJobStatusMessage =
-                    filterTargetJob(jobStatus, clusterJobStatuses);
-
-            if (targetJobStatusMessage.isEmpty()) {
-                LOG.warn("No matching jobs found on the cluster");
-                ifRunningMoveToReconciling(jobStatus, previousJobStatus);
-                onTargetJobNotFound(ctx);
-                return false;
-            } else {
-                updateJobStatus(ctx, targetJobStatusMessage.get());
-            }
-            ReconciliationUtils.checkAndUpdateStableSpec(resource.getStatus());
-            return true;
-        } else {
-            LOG.debug("No jobs found on the cluster");
-            // No jobs found on the cluster, it is possible that the jobmanager is still starting up
-            ifRunningMoveToReconciling(jobStatus, previousJobStatus);
-            onNoJobsFound(ctx);
-            return false;
         }
+        return false;
     }
 
     /**
      * Callback when no matching target job was found on a cluster where jobs were found.
      *
      * @param ctx The Flink resource context.
      */
-    protected abstract void onTargetJobNotFound(FlinkResourceContext<R> ctx);
-
-    /**
-     * Callback when no jobs were found on the cluster.
-     *
-     * @param ctx The Flink resource context.
-     */
-    protected void onNoJobsFound(FlinkResourceContext<R> ctx) {}
+    protected void onTargetJobNotFound(FlinkResourceContext<R> ctx) {
+        ctx.getResource()
+                .getStatus()
+                .getJobStatus()
+                .setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
+        ReconciliationUtils.updateForReconciliationError(
+                ctx, new NotFoundException(JOB_NOT_FOUND_ERR));
+        eventRecorder.triggerEvent(
+                ctx.getResource(),
+                EventRecorder.Type.Warning,
+                EventRecorder.Reason.Missing,
+                EventRecorder.Component.Job,
+                JOB_NOT_FOUND_ERR,
+                ctx.getKubernetesClient());
+    }
 
     /**
      * If we observed the job previously in RUNNING state we move to RECONCILING instead as we are
@@ -139,18 +131,7 @@ private void ifRunningMoveToReconciling(JobStatus jobStatus, String previousJobS
      *
      * @param ctx Resource context.
      */
-    protected abstract void onTimeout(FlinkResourceContext<R> ctx);
-
-    /**
-     * Filter the target job status message by the job list from the cluster.
-     *
-     * @param status the target job status.
-     * @param clusterJobStatuses the candidate cluster jobs.
-     * @return The target job status message. If no matched job found, {@code Optional.empty()} will
-     *     be returned.
-     */
-    protected abstract Optional<JobStatusMessage> filterTargetJob(
-            JobStatus status, List<JobStatusMessage> clusterJobStatuses);
+    protected void onTimeout(FlinkResourceContext<R> ctx) {}
 
     /**
      * Update the status in CR according to the cluster job status.
@@ -161,16 +142,13 @@ protected abstract Optional<JobStatusMessage> filterTargetJob(
     private void updateJobStatus(FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
         var resource = ctx.getResource();
         var jobStatus = resource.getStatus().getJobStatus();
-        var previousJobId = jobStatus.getJobId();
         var previousJobStatus = jobStatus.getState();
 
         jobStatus.setState(clusterJobStatus.getJobState().name());
         jobStatus.setJobName(clusterJobStatus.getJobName());
-        jobStatus.setJobId(clusterJobStatus.getJobId().toHexString());
         jobStatus.setStartTime(String.valueOf(clusterJobStatus.getStartTime()));
 
-        if (jobStatus.getJobId().equals(previousJobId)
-                && jobStatus.getState().equals(previousJobStatus)) {
+        if (jobStatus.getState().equals(previousJobStatus)) {
             LOG.debug("Job status ({}) unchanged", previousJobStatus);
         } else {
             jobStatus.setUpdateTime(String.valueOf(System.currentTimeMillis()));

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/exception/UnknownJobException.java

@@ -19,19 +19,11 @@
 
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
-import org.apache.flink.kubernetes.operator.api.status.JobStatus;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
-import org.apache.flink.kubernetes.operator.exception.UnknownJobException;
 import org.apache.flink.kubernetes.operator.observer.ClusterHealthObserver;
 import org.apache.flink.kubernetes.operator.observer.JobStatusObserver;
 import org.apache.flink.kubernetes.operator.observer.SnapshotObserver;
-import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
 import org.apache.flink.kubernetes.operator.utils.EventRecorder;
-import org.apache.flink.runtime.client.JobStatusMessage;
-
-import java.util.Comparator;
-import java.util.List;
-import java.util.Optional;
 
 import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED;
 
@@ -73,46 +65,5 @@ public ApplicationJobObserver(EventRecorder eventRecorder) {
         public void onTimeout(FlinkResourceContext<FlinkDeployment> ctx) {
             observeJmDeployment(ctx);
         }
-
-        @Override
-        protected Optional<JobStatusMessage> filterTargetJob(
-                JobStatus status, List<JobStatusMessage> clusterJobStatuses) {
-            if (!clusterJobStatuses.isEmpty()) {
-                clusterJobStatuses.sort(
-                        Comparator.comparingLong(JobStatusMessage::getStartTime).reversed());
-                return Optional.of(clusterJobStatuses.get(0));
-            }
-            return Optional.empty();
-        }
-
-        @Override
-        protected void onTargetJobNotFound(FlinkResourceContext<FlinkDeployment> ctx) {
-            // This should never happen for application clusters, there is something
-            // wrong
-            setUnknownJobError(ctx);
-        }
-
-        /**
-         * We found a job on an application cluster that doesn't match the expected job. Trigger
-         * error.
-         *
-         * @param ctx Application deployment context.
-         */
-        private void setUnknownJobError(FlinkResourceContext<FlinkDeployment> ctx) {
-            ctx.getResource()
-                    .getStatus()
-                    .getJobStatus()
-                    .setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
-            String err = "Unrecognized Job for Application deployment";
-            logger.error(err);
-            ReconciliationUtils.updateForReconciliationError(ctx, new UnknownJobException(err));
-            eventRecorder.triggerEvent(
-                    ctx.getResource(),
-                    EventRecorder.Type.Warning,
-                    EventRecorder.Reason.Missing,
-                    EventRecorder.Component.Job,
-                    err,
-                    ctx.getKubernetesClient());
-        }
     }
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java

@@ -36,7 +36,7 @@ public void observeFlinkCluster(FlinkResourceContext<FlinkDeployment> ctx) {
         // Check if session cluster can serve rest calls following our practice in JobObserver
         try {
             logger.debug("Observing session cluster");
-            ctx.getFlinkService().listJobs(ctx.getObserveConfig());
+            ctx.getFlinkService().getClusterInfo(ctx.getObserveConfig());
             var rs = ctx.getResource().getStatus().getReconciliationStatus();
             if (rs.getState() == ReconciliationState.DEPLOYED) {
                 rs.markReconciledSpecAsStable();