[Feature-4518][admin] Support for automatically restarting failed jobs (#4519)

Author: aiwenmo

.github/workflows/backend.yaml

@@ -205,7 +205,7 @@ jobs:
             localhost:5000/dinky/flink:flink
       - name: Init  Env Jar
         run: |
-          mkdir O e2e_test/docker-compose-env/dinky/jars
+          mkdir -p e2e_test/docker-compose-env/dinky/jars
           wget -O e2e_test/docker-compose-env/dinky/mysql-connector-java-8.0.30.jar  https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.30/mysql-connector-java-8.0.30.jar &&
           wget -O e2e_test/docker-compose-env/flink/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar  https://repo1.maven.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar &&
           wget -O e2e_test/docker-compose-env/dinky/javax.ws.rs-api-2.1.1.jar  https://repo1.maven.org/maven2/javax/ws/rs/javax.ws.rs-api/2.1.1/javax.ws.rs-api-2.1.1.jar

dinky-admin/src/main/java/org/dinky/data/dto/TaskDTO.java

@@ -157,6 +157,14 @@ public class TaskDTO extends AbstractStatementDTO {
             notes = "ID of the version associated with the task")
     private Integer versionId;
 
+    @ApiModelProperty(
+            value = "Auto Restart",
+            dataType = "Boolean",
+            example = "false",
+            notes =
+                    "Whether to automatically restart the job from the latest checkpoint when it fails or becomes UNKNOWN")
+    private Boolean autoRestart;
+
     @ApiModelProperty(value = "Enabled", required = true, dataType = "Boolean", example = "true")
     private Boolean enabled;
 

dinky-admin/src/main/java/org/dinky/data/dto/TaskSaveDTO.java

@@ -165,6 +165,13 @@ public class TaskSaveDTO {
     @TableField(typeHandler = ListTypeHandler.class)
     private List<Integer> secondLevelOwners;
 
+    @ApiModelProperty(
+            value = "Auto Restart",
+            dataType = "Boolean",
+            example = "false",
+            notes = "Auto Restart option for the task")
+    private Boolean autoRestart;
+
     public Task toTaskEntity() {
         Task task = new Task();
         BeanUtil.copyProperties(this, task);

dinky-admin/src/main/java/org/dinky/data/model/Task.java

@@ -161,6 +161,14 @@ public class Task extends SuperEntity<Task> {
             notes = "ID of the version associated with the task")
     private Integer versionId;
 
+    @ApiModelProperty(
+            value = "Auto Restart",
+            dataType = "Boolean",
+            example = "false",
+            notes =
+                    "Whether to automatically restart the job from the latest checkpoint when it fails or becomes UNKNOWN")
+    private Boolean autoRestart;
+
     @ApiModelProperty(value = "Enabled", dataType = "Boolean", example = "true", notes = "Whether the task is enabled")
     private Boolean enabled;
 

dinky-admin/src/main/java/org/dinky/job/handler/JobRefreshHandler.java

@@ -41,6 +41,7 @@
 import org.dinky.data.model.ClusterInstance;
 import org.dinky.data.model.SystemConfiguration;
 import org.dinky.data.model.ext.JobInfoDetail;
+import org.dinky.data.model.job.History;
 import org.dinky.data.model.job.JobInstance;
 import org.dinky.gateway.Gateway;
 import org.dinky.gateway.config.GatewayConfig;
@@ -52,6 +53,7 @@
 import org.dinky.service.HistoryService;
 import org.dinky.service.JobHistoryService;
 import org.dinky.service.JobInstanceService;
+import org.dinky.service.TaskService;
 import org.dinky.utils.JsonUtils;
 import org.dinky.utils.TimeUtil;
 
@@ -83,12 +85,14 @@ public class JobRefreshHandler {
     private static final JobHistoryService jobHistoryService;
     private static final ClusterInstanceService clusterInstanceService;
     private static final HistoryService historyService;
+    private static final TaskService taskService;
 
     static {
         jobInstanceService = SpringContextUtils.getBean("jobInstanceServiceImpl", JobInstanceService.class);
         jobHistoryService = SpringContextUtils.getBean("jobHistoryServiceImpl", JobHistoryService.class);
         clusterInstanceService = SpringContextUtils.getBean("clusterInstanceServiceImpl", ClusterInstanceService.class);
         historyService = SpringContextUtils.getBean("historyServiceImpl", HistoryService.class);
+        taskService = SpringContextUtils.getBean("taskServiceImpl", TaskService.class);
     }
 
     /**
@@ -170,7 +174,9 @@ public static boolean refreshJob(JobInfoDetail jobInfoDetail, boolean needSave)
 
         boolean isTransition = false;
 
-        if (JobStatus.isTransition(jobInstance.getStatus())) {
+        if (JobStatus.isTransition(
+                jobInstance.getStatus(),
+                Asserts.isNull(jobDataDto.getJob()) ? null : jobDataDto.getJob().getEndTime())) {
             Long finishTime = TimeUtil.localDateTimeToLong(jobInstance.getFinishTime());
             long duration = Duration.between(jobInstance.getFinishTime(), LocalDateTime.now())
                     .toMinutes();
@@ -217,6 +223,10 @@ public static boolean refreshJob(JobInfoDetail jobInfoDetail, boolean needSave)
         if (isDone) {
             try {
                 log.debug("Job is done: {}->{}", jobInstance.getId(), jobInstance.getName());
+                // 检查是否需要自动重启
+                if (shouldAutoRestart(jobInstance, jobInfoDetail)) {
+                    tryAutoRestart(jobInstance, jobInfoDetail);
+                }
                 handleJobDone(jobInfoDetail);
             } catch (Exception e) {
                 log.error("failed handel job done：", e);
@@ -346,6 +356,94 @@ private static void handleJobDone(JobInfoDetail jobInfoDetail) {
         }
     }
 
+    /**
+     * Check if the job should be auto-restarted.
+     *
+     * @param jobInstance The job instance.
+     * @param jobInfoDetail The job info detail.
+     * @return True if the job should be auto-restarted, false otherwise.
+     */
+    private static boolean shouldAutoRestart(JobInstance jobInstance, JobInfoDetail jobInfoDetail) {
+        String status = jobInstance.getStatus();
+        // 只对FAILED和UNKNOWN状态进行自动重启
+        if (!JobStatus.FAILED.getValue().equals(status)
+                && !JobStatus.UNKNOWN.getValue().equals(status)) {
+            return false;
+        }
+
+        // 检查任务配置中是否启用了自动重启
+        try {
+            History history = jobInfoDetail.getHistory();
+            if (Asserts.isNull(history) || Asserts.isNull(history.getConfigJson())) {
+                return false;
+            }
+
+            JobConfig jobConfig = history.getConfigJson();
+            Boolean autoRestart = jobConfig.getAutoRestart();
+            return Boolean.TRUE.equals(autoRestart);
+        } catch (Exception e) {
+            log.warn("Failed to check auto restart config for job {}: {}", jobInstance.getId(), e.getMessage());
+            return false;
+        }
+    }
+
+    /**
+     * Try to auto-restart the job from the latest checkpoint.
+     *
+     * @param jobInstance The job instance.
+     * @param jobInfoDetail The job info detail.
+     */
+    private static void tryAutoRestart(JobInstance jobInstance, JobInfoDetail jobInfoDetail) {
+        if (Asserts.isNull(jobInstance.getTaskId())) {
+            log.warn("Cannot auto restart job {}: taskId is null", jobInstance.getId());
+            return;
+        }
+
+        try {
+            // 获取最新的checkpoint路径
+            String checkpointPath = getLatestCheckpointPath(jobInfoDetail.getJobDataDto());
+            log.info("Auto restarting job {} from checkpoint: {}", jobInstance.getId(), checkpointPath);
+            taskService.restartTask(jobInstance.getTaskId(), checkpointPath);
+            log.info("Auto restart job {} triggered successfully", jobInstance.getId());
+        } catch (Exception e) {
+            log.error("Failed to auto restart job {}: {}", jobInstance.getId(), e.getMessage(), e);
+        }
+    }
+
+    /**
+     * Get the latest checkpoint path from JobDataDto.
+     *
+     * @param jobDataDto The job data DTO.
+     * @return The latest checkpoint path, or null if not found.
+     */
+    private static String getLatestCheckpointPath(JobDataDto jobDataDto) {
+        if (Asserts.isNull(jobDataDto) || Asserts.isNull(jobDataDto.getCheckpoints())) {
+            return null;
+        }
+
+        CheckPointOverView checkpoints = jobDataDto.getCheckpoints();
+        CheckPointOverView.LatestCheckpoints latestCheckpoints = checkpoints.getLatestCheckpoints();
+        if (Asserts.isNull(latestCheckpoints)) {
+            return null;
+        }
+
+        // 优先使用completed checkpoint
+        CheckPointOverView.CompletedCheckpointStatistics completedCheckpoint =
+                latestCheckpoints.getCompletedCheckpointStatistics();
+        if (Asserts.isNotNull(completedCheckpoint) && Asserts.isNotNullString(completedCheckpoint.getExternalPath())) {
+            return completedCheckpoint.getExternalPath();
+        }
+
+        // 如果没有completed checkpoint，尝试使用savepoint
+        CheckPointOverView.CompletedCheckpointStatistics savepointStatistics =
+                latestCheckpoints.getSavepointStatistics();
+        if (Asserts.isNotNull(savepointStatistics) && Asserts.isNotNullString(savepointStatistics.getExternalPath())) {
+            return savepointStatistics.getExternalPath();
+        }
+
+        return null;
+    }
+
     /**
      * In a YARN cluster with HA mode enabled,
      * if the jobManagerHost cannot be connected,

dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java

@@ -326,8 +326,8 @@ public String buildEnvSql(AbstractStatementDTO task) {
                 sql += envTask.getStatement() + CommonConstant.LineSep;
             }
         }
-        log.info("Initializing data permissions...");
-        userService.buildRowPermission();
+        /*log.info("Initializing data permissions...");
+        userService.buildRowPermission();*/
         log.info("Finish initialize FlinkSQLEnv.");
         return sql;
     }

dinky-admin/src/main/resources/db/migration/h2/V20250301.1.3.0__release.sql

@@ -38,4 +38,6 @@ insert into `dinky_sys_menu` (`id`, `parent_id`, `name`, `path`, `component`, `p
 values (178, 177, '编辑', '/settings/globalsetting/approval/edit', null, 'settings:globalsetting:approval:edit',
         'EditOutlined', 'F', 0, 171, '2024-12-30 23:45:30', '2024-12-30 23:45:30', null);
 
+ALTER TABLE `dinky_task` ADD COLUMN `auto_restart` tinyint(1) DEFAULT 0 COMMENT '是否自动重启，当作业失败或UNKNOWN时自动从最新checkpoint重启';
+
 SET FOREIGN_KEY_CHECKS = 1;

dinky-admin/src/main/resources/db/migration/mysql/V20250301.1.3.0__release.sql

@@ -70,5 +70,7 @@ insert into `dinky_sys_menu` (`id`, `parent_id`, `name`, `path`, `component`, `p
 values (178, 177, '编辑', '/settings/globalsetting/approval/edit', null, 'settings:globalsetting:approval:edit',
         'EditOutlined', 'F', 0, 171, '2024-12-30 23:45:30', '2024-12-30 23:45:30', null);
 
+ALTER TABLE `dinky_task` ADD COLUMN `auto_restart` tinyint(1) NULL DEFAULT 0 COMMENT '是否自动重启，当作业失败或UNKNOWN时自动从最新checkpoint重启';
+
 SET
     FOREIGN_KEY_CHECKS = 1;

dinky-admin/src/main/resources/db/migration/postgresql/V20250301.1.3.0__release.sql

@@ -79,3 +79,6 @@ insert into public.dinky_sys_menu (id, parent_id, name, path, component, perms,
                               order_num, create_time, update_time, note)
 values (178, 177, '编辑', '/settings/globalsetting/approval/edit', null, 'settings:globalsetting:approval:edit',
         'EditOutlined', 'F', 0, 171, '2024-12-30 23:45:30', '2024-12-30 23:45:30', null);
+
+ALTER TABLE public.dinky_task ADD COLUMN IF NOT EXISTS auto_restart boolean DEFAULT FALSE;
+COMMENT ON COLUMN public.dinky_task.auto_restart IS '是否自动重启，当作业失败或UNKNOWN时自动从最新checkpoint重启';

dinky-common/src/main/java/org/dinky/data/enums/JobStatus.java

@@ -109,9 +109,10 @@ public static boolean isDone(String value) {
         }
     }
 
-    public static boolean isTransition(String value) {
+    public static boolean isTransition(String value, Long endTime) {
         switch (get(value)) {
             case FAILED:
+                return endTime == -1;
             case RECONNECTING:
                 return true;
             default: