fix: Add jobFlowStatus parameter to queryLastJobRuns method

Author: itinycheng

flink-platform-dao/src/main/java/com/flink/platform/dao/mapper/JobRunInfoMapper.java

@@ -2,6 +2,7 @@
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.flink.platform.common.enums.ExecutionStatus;
+import com.flink.platform.common.enums.JobFlowStatus;
 import com.flink.platform.common.enums.JobFlowType;
 import com.flink.platform.common.enums.JobStatus;
 import com.flink.platform.dao.entity.JobRunInfo;
@@ -63,6 +64,9 @@ SELECT max(r.id) as run_id
                             <if test="jobFlowType != null">
                                 AND f.`type` = #{jobFlowType}
                             </if>
+                            <if test="jobFlowStatus != null">
+                                AND f.`status` = #{jobFlowStatus}
+                            </if>
                             <if test="jobStatus != null">
                                 AND j.status = #{jobStatus}
                             </if>
@@ -79,6 +83,7 @@ SELECT max(r.id) as run_id
             """)
     List<JobRunInfo> queryLastJobRuns(
             @Param("jobFlowType") JobFlowType jobFlowType,
+            @Param("jobFlowStatus") JobFlowStatus jobFlowStatus,
             @Param("jobStatus") JobStatus jobStatus,
             @Param("jobRunStatusList") List<ExecutionStatus> jobRunStatusList);
 }

flink-platform-dao/src/main/java/com/flink/platform/dao/service/JobRunInfoService.java

@@ -3,6 +3,7 @@
 import com.baomidou.dynamic.datasource.annotation.DS;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.flink.platform.common.enums.JobFlowStatus;
 import com.flink.platform.common.enums.JobFlowType;
 import com.flink.platform.common.enums.JobStatus;
 import com.flink.platform.dao.entity.JobRunInfo;
@@ -33,6 +34,7 @@ public JobRunInfo findRunningJob(@Nonnull Long jobId) {
     }
 
     public List<JobRunInfo> getJobRunsWithUnexpectedStatus() {
-        return this.baseMapper.queryLastJobRuns(JobFlowType.JOB_LIST, JobStatus.ONLINE, UNEXPECTED);
+        return this.baseMapper.queryLastJobRuns(
+                JobFlowType.JOB_LIST, JobFlowStatus.SCHEDULING, JobStatus.ONLINE, UNEXPECTED);
     }
 }