[BugFix] Make TaskRun's status be compatible with lower version (backport #60438) (backport #60440) (#60447)

Signed-off-by: shuming.li <ming.moriarty@gmail.com>
Co-authored-by: shuming.li <ming.moriarty@gmail.com>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/scheduler/TaskManager.java

@@ -592,6 +592,13 @@ public void saveTasksV2(ImageWriter imageWriter) throws IOException, SRMetaBlock
 
         writer.writeInt(runStatusList.size());
         for (TaskRunStatus status : runStatusList) {
+            // TODO: This is to be compatible with the old version of TaskRunStatus when degraded from a higher version
+            // because SKIPPED state is not defined in the lower version.
+            // NOTE: This can be removed in the next 3.4 version release.
+            if (status.getState() != null && status.getState().equals(Constants.TaskRunState.SKIPPED)) {
+                status.setState(Constants.TaskRunState.SUCCESS);
+                LOG.warn("TaskRunStatus state is SKIPPED, change to SUCCESS, status: {}", status);
+            }
             writer.writeJson(status);
         }
 
@@ -683,6 +690,22 @@ private boolean isSameTaskRunJob(TaskRunStatus taskRunStatus,
     }
 
     public void replayCreateTaskRun(TaskRunStatus status) {
+        try {
+            doReplayCreateTaskRun(status);
+        } catch (Exception e) {
+            LOG.warn("replay create task run failed, status: {}, error: {}", status, e.getMessage());
+            // The task run will be replayed in FE restart, If the replay fails, it will cause FE restart failed.
+            // It's fine to discard the task run since it's only task's history records and can be retried later.
+        }
+    }
+
+    private void doReplayCreateTaskRun(TaskRunStatus status) {
+        // NOTE: If current FE is downgraded from a higher version and TaskRunStatus#State is new added which is not defined
+        // in current version, status.getState() will be null.
+        if (status == null || status.getState() == null || Strings.isNullOrEmpty(status.getTaskName())) {
+            LOG.warn("replayCreateTaskRun: status is null or taskId is invalid, status: {}", status);
+            return;
+        }
         if (status.getState().isFinishState() && System.currentTimeMillis() > status.getExpireTime()) {
             return;
         }

fe/fe-core/src/test/java/com/starrocks/scheduler/TaskManagerTest.java

@@ -26,7 +26,10 @@
 import com.starrocks.common.util.ThreadUtil;
 import com.starrocks.common.util.TimeUtils;
 import com.starrocks.common.util.UUIDUtil;
+import com.starrocks.persist.ImageWriter;
+import com.starrocks.persist.metablock.SRMetaBlockReader;
 import com.starrocks.qe.ConnectContext;
+import com.starrocks.scheduler.history.TaskRunHistory;
 import com.starrocks.scheduler.persist.TaskRunStatus;
 import com.starrocks.scheduler.persist.TaskRunStatusChange;
 import com.starrocks.server.GlobalStateMgr;
@@ -886,4 +889,49 @@ public void testTaskRunWithLargeDefinition3() {
         String definition = taskRun1.getStatus().getDefinition();
         Assert.assertTrue(definition == null);
     }
+
+    @Test
+    public void saveTasksV2SkipsSkippedTaskRunStatuses() throws Exception {
+        UtFrameUtils.PseudoImage image = new UtFrameUtils.PseudoImage();
+        {
+            TaskManager taskManager = new TaskManager();
+            ImageWriter imageWriter = image.getImageWriter();
+
+            Task task = new Task("task");
+            task.setId(1L);
+            taskManager.replayCreateTask(task);
+
+            TaskRunStatus skippedStatus = new TaskRunStatus();
+            skippedStatus.setTaskId(1);
+            skippedStatus.setQueryId("task_run_1");
+            skippedStatus.setTaskName("task_run_1");
+            skippedStatus.setState(Constants.TaskRunState.SKIPPED);
+            skippedStatus.setExpireTime(System.currentTimeMillis() + 1000000);
+            taskManager.replayCreateTaskRun(skippedStatus);
+
+            TaskRunStatus validStatus = new TaskRunStatus();
+            validStatus.setTaskId(2);
+            validStatus.setQueryId("task_run_2");
+            validStatus.setTaskName("task_run_2");
+            validStatus.setState(Constants.TaskRunState.SUCCESS);
+            validStatus.setExpireTime(System.currentTimeMillis() + 1000000);
+            taskManager.replayCreateTaskRun(validStatus);
+
+            TaskRunHistory taskRunHistory = taskManager.getTaskRunHistory();
+            Assert.assertEquals(2, taskRunHistory.getTaskRunCount());
+
+            taskManager.saveTasksV2(imageWriter);
+        }
+
+        SRMetaBlockReader imageReader = image.getMetaBlockReader();
+        {
+            TaskManager taskManager = new TaskManager();
+            taskManager.loadTasksV2(imageReader);
+            TaskRunHistory taskRunHistory = taskManager.getTaskRunHistory();
+            Assert.assertEquals(2, taskRunHistory.getTaskRunCount());
+            taskRunHistory.getInMemoryHistory()
+                    .stream()
+                    .forEach(status -> Assert.assertEquals(status.getState(), Constants.TaskRunState.SUCCESS));
+        }
+    }
 }