branch-4.1: [fix](job scheduler) Record task finish time on execution failure #66232

fe/fe-core/src/main/java/org/apache/doris/job/task/AbstractTask.java

@@ -71,6 +71,7 @@ public boolean onFail() throws JobException {
             return false;
         }
         status = TaskStatus.FAILED;
+        setFinishTimeMs(System.currentTimeMillis());
         if (!isCallable()) {
             return false;
         }

fe/fe-core/src/test/java/org/apache/doris/job/base/AbstractJobStatusTest.java

@@ -66,6 +66,18 @@ boolean cancelLogicCalled() {
         }
     }
 
+    private static class FailingTask extends DummyTask {
+        FailingTask(long taskId) {
+            super(taskId);
+            setJobId(1L);
+        }
+
+        @Override
+        public void run() throws JobException {
+            throw new JobException("task failed");
+        }
+    }
+
     private static class DummyJob extends AbstractJob<DummyTask, Void> {
         private final List<DummyTask> history = new ArrayList<>();
 
@@ -131,6 +143,17 @@ void testPendingFromPaused() throws Exception {
         Assertions.assertEquals(JobStatus.PENDING, job.getJobStatus());
     }
 
+    @Test
+    void testTaskFailureSetsFinishTime() throws Exception {
+        FailingTask task = new FailingTask(100L);
+
+        task.runTask();
+
+        Assertions.assertEquals(TaskStatus.FAILED, task.getStatus());
+        Assertions.assertNotNull(task.getFinishTimeMs());
+        Assertions.assertEquals("task failed", task.getErrMsg());
+    }
+
     @Test
     void testPendingFromRunning() throws Exception {
         DummyJob job = new DummyJob(JobStatus.RUNNING);