fix: Alert task error

Author: itinycheng

docs/sql/addition.sql

@@ -17,3 +17,8 @@ CREATE INDEX t_job_run_job_id_idx USING BTREE ON platform.t_job_run (job_id);
 
 CREATE INDEX t_job_flow_run_name_idx USING BTREE ON platform.t_job_flow_run (name);
 CREATE INDEX t_job_run_name_idx USING BTREE ON platform.t_job_run (name);
+
+-- 2024-12-04
+CREATE INDEX t_job_run_stop_time_idx USING BTREE ON platform.t_job_run (stop_time);
+CREATE INDEX t_job_flow_run_end_time_idx USING BTREE ON platform.t_job_flow_run (end_time);
+

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

@@ -66,15 +66,15 @@ SELECT max(r.id) as run_id
                             <if test="jobStatus != null">
                                 AND j.status = #{jobStatus}
                             </if>
-                            <if test="jobRunStatusList != null and jobRunStatusList.size() > 0">
-                                AND r.status in
-                                <foreach collection="jobRunStatusList" item="jobRunStatus" open='(' close=')' separator=','>
-                                    #{jobRunStatus}
-                                </foreach>
-                            </if>
                         GROUP BY r.job_id
                     ) t2
-                WHERE t1.id = t2.run_id;
+                WHERE t1.id = t2.run_id
+                <if test="jobRunStatusList != null and jobRunStatusList.size() > 0">
+                    AND t1.status in
+                    <foreach collection="jobRunStatusList" item="jobRunStatus" open='(' close=')' separator=','>
+                        #{jobRunStatus}
+                    </foreach>
+                </if>
             </script>
             """)
     List<JobRunInfo> queryLastJobRuns(

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

@@ -14,10 +14,10 @@
 @DS("master_platform")
 public class WorkerService extends ServiceImpl<WorkerMapper, Worker> {
 
-    public Worker getCurrentWorker() {
+    public Worker getCurWorkerIdAndRole() {
         return getOne(new QueryWrapper<Worker>()
                 .lambda()
-                .select(Worker::getId)
+                .select(Worker::getId, Worker::getRole)
                 .eq(Worker::getIp, HOST_IP)
                 .last("LIMIT 1"));
     }

flink-platform-monitor/src/main/java/com/flink/platform/monitor/JobsInJobListStatusChecker.java

@@ -32,7 +32,7 @@ public class JobsInJobListStatusChecker {
 
     @Scheduled(initialDelay = 2 * 60 * 1000, fixedDelay = 10 * 60 * 1000)
     public void checkJobStatus() {
-        Worker worker = workerService.getCurrentWorker();
+        Worker worker = workerService.getCurWorkerIdAndRole();
         if (worker == null || !LEADER.equals(worker.getRole())) {
             log.info("Current worker is not leader, skip checkJobStatus.");
             return;

flink-platform-monitor/src/main/java/com/flink/platform/monitor/UnscheduledJobFlowChecker.java

@@ -28,7 +28,7 @@ public class UnscheduledJobFlowChecker {
 
     @Scheduled(initialDelay = 2 * 60 * 1000, fixedDelay = 60 * 60 * 1000)
     public void checkUnscheduledWorkflow() {
-        Worker worker = workerService.getCurrentWorker();
+        Worker worker = workerService.getCurWorkerIdAndRole();
         if (worker == null || !LEADER.equals(worker.getRole())) {
             log.info("Current worker is not leader, skip checkUnscheduledWorkflow.");
             return;

flink-platform-ui/frontend

@@ -68,7 +68,7 @@ public void initHeartbeat() {
 
     public synchronized void heartbeat() {
         // 1. Update worker heartbeat info.
-        Worker worker = workerService.getCurrentWorker();
+        Worker worker = workerService.getCurWorkerIdAndRole();
         Long workerId = worker != null ? worker.getId() : null;
 
         worker = new Worker();