[Improve][SeaTunnel-Web] Change JobStatus to enum type to avoid hard …

…coding (#205)
apache · Sep 7, 2024 · 94d8a39 · 94d8a39
1 parent 528746d
commit 94d8a39
Show file tree

Hide file tree

Showing 18 changed files with 88 additions and 66 deletions.
diff --git a/...l-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/JobInstance.java b/...l-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/JobInstance.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.app.dal.entity;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
@@ -42,7 +44,7 @@ public class JobInstance {
     private Long jobDefineId;
 
     @TableField("job_status")
-    private String jobStatus;
+    private JobStatus jobStatus;
 
     @TableField("job_config")
     private String jobConfig;

diff --git a/...el-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/JobMetrics.java b/...el-server/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/JobMetrics.java
@@ -16,6 +16,8 @@
  */
 package org.apache.seatunnel.app.dal.entity;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
@@ -65,7 +67,7 @@ public class JobMetrics {
     private long recordDelay;
 
     @TableField("status")
-    private String status;
+    private JobStatus status;
 
     @TableField("create_user_id")
     private Integer createUserId;

diff --git a/...p/src/main/java/org/apache/seatunnel/app/domain/response/executor/JobExecutionStatus.java b/...p/src/main/java/org/apache/seatunnel/app/domain/response/executor/JobExecutionStatus.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.app.domain.response.executor;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -26,7 +28,7 @@
 @NoArgsConstructor
 public class JobExecutionStatus {
 
-    private String jobStatus;
+    private JobStatus jobStatus;
 
     private String errorMessage;
 }
diff --git a/...in/java/org/apache/seatunnel/app/domain/response/metrics/JobPipelineDetailMetricsRes.java b/...in/java/org/apache/seatunnel/app/domain/response/metrics/JobPipelineDetailMetricsRes.java
@@ -16,6 +16,8 @@
  */
 package org.apache.seatunnel.app.domain.response.metrics;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -45,5 +47,5 @@ public class JobPipelineDetailMetricsRes {
 
     private long recordDelay;
 
-    private String status;
+    private JobStatus status;
 }
diff --git a/...n/java/org/apache/seatunnel/app/domain/response/metrics/JobPipelineSummaryMetricsRes.java b/...n/java/org/apache/seatunnel/app/domain/response/metrics/JobPipelineSummaryMetricsRes.java
@@ -16,6 +16,8 @@
  */
 package org.apache.seatunnel.app.domain.response.metrics;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
@@ -28,5 +30,5 @@ public class JobPipelineSummaryMetricsRes {
 
     private long writeRowCount;
 
-    private String status;
+    private JobStatus status;
 }
diff --git a/.../src/main/java/org/apache/seatunnel/app/domain/response/metrics/JobSummaryMetricsRes.java b/.../src/main/java/org/apache/seatunnel/app/domain/response/metrics/JobSummaryMetricsRes.java
@@ -16,6 +16,8 @@
  */
 package org.apache.seatunnel.app.domain.response.metrics;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
@@ -30,5 +32,5 @@ public class JobSummaryMetricsRes {
 
     private long writeRowCount;
 
-    private String status;
+    private JobStatus status;
 }
diff --git a/...nel-app/src/main/java/org/apache/seatunnel/app/domain/response/task/JobSimpleInfoRes.java b/...nel-app/src/main/java/org/apache/seatunnel/app/domain/response/task/JobSimpleInfoRes.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.app.domain.response.task;
 
+import org.apache.seatunnel.engine.core.job.JobStatus;
+
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
@@ -40,7 +42,7 @@ public class JobSimpleInfoRes {
     private String jobName;
 
     @ApiModelProperty(value = "job status", dataType = "String")
-    private String jobStatus;
+    private JobStatus jobStatus;
 
     @ApiModelProperty(value = "job plan", dataType = "String")
     private String jobPlan;

diff --git a/...unnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobExecutorServiceImpl.java b/...unnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobExecutorServiceImpl.java
@@ -55,7 +55,6 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.util.Date;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -128,7 +127,7 @@ private void executeJobBySeaTunnel(Integer userId, String filePath, Long jobInst
         } catch (Throwable e) {
             log.error("Job execution submission failed.", e);
             JobInstance jobInstance = jobInstanceDao.getJobInstance(jobInstanceId);
-            jobInstance.setJobStatus(JobStatus.FAILED.name());
+            jobInstance.setJobStatus(JobStatus.FAILED);
             jobInstance.setEndTime(new Date());
             String jobInstanceErrorMessage = JobUtils.getJobInstanceErrorMessage(e.getMessage());
             jobInstance.setErrorMessage(jobInstanceErrorMessage);
@@ -183,14 +182,14 @@ private SeaTunnelClient createSeaTunnelClient() {
     @Override
     public Result<Void> jobPause(Integer userId, Long jobInstanceId) {
         JobInstance jobInstance = jobInstanceDao.getJobInstance(jobInstanceId);
-        if (Objects.equals(
-                getJobStatusFromEngine(jobInstance, jobInstance.getJobEngineId()), "RUNNING")) {
+        if (getJobStatusFromEngine(jobInstance, jobInstance.getJobEngineId())
+                == JobStatus.RUNNING) {
             pauseJobInEngine(jobInstance.getJobEngineId());
         }
         return Result.success();
     }
 
-    private String getJobStatusFromEngine(@NonNull JobInstance jobInstance, String jobEngineId) {
+    private JobStatus getJobStatusFromEngine(@NonNull JobInstance jobInstance, String jobEngineId) {
 
         Engine engine = new Engine(jobInstance.getEngineName(), jobInstance.getEngineVersion());
 

diff --git a/...unnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobInstanceServiceImpl.java b/...unnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobInstanceServiceImpl.java
@@ -366,7 +366,7 @@ public void complete(
         funcPermissionCheck(SeatunnelFuncPermissionKeyConstant.JOB_EXECUTOR_COMPLETE, userId);
         JobInstance jobInstance = jobInstanceDao.getJobInstanceMapper().selectById(jobInstanceId);
         jobMetricsService.syncJobDataToDb(jobInstance, userId, jobEngineId);
-        jobInstance.setJobStatus(jobResult.getStatus().name());
+        jobInstance.setJobStatus(jobResult.getStatus());
         jobInstance.setJobEngineId(jobEngineId);
         jobInstance.setUpdateUserId(userId);
         jobInstance.setErrorMessage(JobUtils.getJobInstanceErrorMessage(jobResult.getError()));

diff --git a/...tunnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobMetricsServiceImpl.java b/...tunnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobMetricsServiceImpl.java
@@ -33,6 +33,7 @@
 import org.apache.seatunnel.app.thirdparty.metrics.EngineMetricsExtractorFactory;
 import org.apache.seatunnel.app.thirdparty.metrics.IEngineMetricsExtractor;
 import org.apache.seatunnel.app.utils.JobUtils;
+import org.apache.seatunnel.engine.core.job.JobStatus;
 import org.apache.seatunnel.server.common.CodeGenerateUtils;
 import org.apache.seatunnel.server.common.Constants;
 import org.apache.seatunnel.server.common.SeatunnelErrorEnum;
@@ -89,7 +90,7 @@ public JobSummaryMetricsRes getJobSummaryMetrics(
         Engine engine = new Engine(jobInstance.getEngineName(), jobInstance.getEngineVersion());
         IEngineMetricsExtractor engineMetricsExtractor =
                 (new EngineMetricsExtractorFactory(engine)).getEngineMetricsExtractor();
-        String jobStatus = engineMetricsExtractor.getJobStatus(jobEngineId);
+        JobStatus jobStatus = engineMetricsExtractor.getJobStatus(jobEngineId);
 
         List<JobMetrics> jobPipelineDetailMetrics =
                 getJobPipelineDetailMetrics(jobInstance, userId);
@@ -160,8 +161,8 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsBatch(
             log.info("jobEngineId={}", jobInstance.getJobEngineId());
 
             if (jobInstance.getJobStatus() == null
-                    || jobInstance.getJobStatus().equals("FAILED")
-                    || jobInstance.getJobStatus().equals("RUNNING")) {
+                    || jobInstance.getJobStatus() == JobStatus.FAILED
+                    || jobInstance.getJobStatus() == JobStatus.RUNNING) {
                 // Obtain monitoring information from the collection of running jobs returned from
                 // the engine
                 if (!allRunningJobMetricsFromEngine.isEmpty()
@@ -192,15 +193,15 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsBatch(
                     if (jobMetricsFromDb != null) {
                         jobSummaryMetricsResMap.put(jobInstance.getId(), jobMetricsFromDb);
                     }
-                    if ("RUNNING".equals(jobInstance.getJobStatus())) {
+                    if (jobInstance.getJobStatus() == JobStatus.RUNNING) {
                         // Set the job status of jobInstance and jobMetrics in the database to
                         // finished
-                        jobInstance.setJobStatus("FINISHED");
+                        jobInstance.setJobStatus(JobStatus.FINISHED);
                         jobInstanceDao.getJobInstanceMapper().updateById(jobInstance);
                     }
                 }
-            } else if (jobInstance.getJobStatus().equals("FINISHED")
-                    || jobInstance.getJobStatus().equals("CANCELED")) {
+            } else if (jobInstance.getJobStatus() == JobStatus.FINISHED
+                    || jobInstance.getJobStatus() == JobStatus.CANCELED) {
                 // If the status of the job is finished or cancelled, the monitoring information is
                 // directly obtained from MySQL
                 JobSummaryMetricsRes jobMetricsFromDb =
@@ -222,7 +223,7 @@ private void modifyAndUpdateJobInstanceAndJobMetrics(
             Map<Long, HashMap<Integer, JobMetrics>> allRunningJobMetricsFromEngine,
             Map<Long, Long> jobInstanceIdAndJobEngineIdMap,
             Integer userId) {
-        jobInstance.setJobStatus("RUNNING");
+        jobInstance.setJobStatus(JobStatus.RUNNING);
         HashMap<Integer, JobMetrics> jobMetricsFromEngine =
                 allRunningJobMetricsFromEngine.get(
                         jobInstanceIdAndJobEngineIdMap.get(jobInstance.getId()));
@@ -246,7 +247,7 @@ private void modifyAndUpdateJobInstanceAndJobMetrics(
                                     jobMetricsFromEngine
                                             .get(jobMetrics.getPipelineId())
                                             .getWriteRowCount()));
-            jobMetricsFromDb.forEach(jobMetrics -> jobMetrics.setStatus("RUNNING"));
+            jobMetricsFromDb.forEach(jobMetrics -> jobMetrics.setStatus(JobStatus.RUNNING));
 
             updateJobInstanceAndMetrics(jobInstance, jobMetricsFromDb);
         }
@@ -265,7 +266,7 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsStreaming(
 
             try {
                 if (jobInstance.getJobStatus() != null
-                        && jobInstance.getJobStatus().equals("CANCELED")) {
+                        && jobInstance.getJobStatus() == JobStatus.CANCELED) {
                     // If the status of the job is finished or cancelled
                     // the monitoring information is directly obtained from MySQL
                     JobSummaryMetricsRes jobMetricsFromDb =
@@ -278,8 +279,8 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsStreaming(
                     jobSummaryMetricsResMap.put(jobInstance.getId(), jobMetricsFromDb);
 
                 } else if (jobInstance.getJobStatus() != null
-                        && (jobInstance.getJobStatus().equals("FINISHED")
-                                || jobInstance.getJobStatus().equals("FAILED"))) {
+                        && (jobInstance.getJobStatus() == JobStatus.FINISHED
+                                || jobInstance.getJobStatus() == JobStatus.FAILED)) {
                     // Obtain monitoring information from the collection of running jobs returned
                     // from
                     // the engine
@@ -332,9 +333,9 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsStreaming(
                                         jobInstance);
                         jobSummaryMetricsResMap.put(jobInstance.getId(), jobMetricsFromEngineRes);
                     } else {
-                        String jobStatusByJobEngineId = null;
+                        JobStatus jobStatus = null;
                         try {
-                            jobStatusByJobEngineId =
+                            jobStatus =
                                     getJobStatusByJobEngineId(
                                             String.valueOf(
                                                     jobInstanceIdAndJobEngineIdMap.get(
@@ -345,8 +346,8 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsStreaming(
                                     jobInstance.getId());
                         }
 
-                        if (jobStatusByJobEngineId != null) {
-                            jobInstance.setJobStatus(jobStatusByJobEngineId);
+                        if (jobStatus != null) {
+                            jobInstance.setJobStatus(jobStatus);
                             jobInstanceDao.update(jobInstance);
                             JobSummaryMetricsRes jobSummaryMetricsResByDb =
                                     getJobSummaryMetricsResByDb(
@@ -365,7 +366,7 @@ private Map<Long, JobSummaryMetricsRes> getMatricsListIfTaskTypeIsStreaming(
                                                     jobInstanceIdAndJobEngineIdMap.get(
                                                             jobInstance.getId())));
                             if (!jobMetricsFromDb.isEmpty()) {
-                                String finalJobStatusByJobEngineId = jobStatusByJobEngineId;
+                                JobStatus finalJobStatusByJobEngineId = jobStatus;
                                 jobMetricsFromDb.forEach(
                                         jobMetrics ->
                                                 jobMetrics.setStatus(finalJobStatusByJobEngineId));
@@ -402,7 +403,8 @@ private JobSummaryMetricsRes getRunningJobMetricsFromEngine(
 
         log.info("jobInstance={}", jobInstance);
 
-        return new JobSummaryMetricsRes(jobInstance.getId(), 1L, readCount, writeCount, "RUNNING");
+        return new JobSummaryMetricsRes(
+                jobInstance.getId(), 1L, readCount, writeCount, JobStatus.RUNNING);
     }
 
     private JobSummaryMetricsRes getJobSummaryMetricsResByDb(
@@ -442,7 +444,7 @@ private void updateJobInstanceAndMetrics(JobInstance jobInstance, List<JobMetric
         }
     }
 
-    private String getJobStatusByJobEngineId(String jobEngineId) {
+    private JobStatus getJobStatusByJobEngineId(String jobEngineId) {
         return SeaTunnelEngineProxy.getInstance().getJobStatus(jobEngineId);
     }
 
@@ -581,7 +583,7 @@ private void syncMetricsToDb(
                 jobMetricsDao.getJobMetricsMapper().insertBatchMetrics(jobMetricsFromEngine);
             }
         } else {
-            String jobStatus = getJobStatusByJobEngineId(jobEngineId);
+            JobStatus jobStatus = getJobStatusByJobEngineId(jobEngineId);
             for (JobMetrics jobMetrics : jobMetricsFromDb) {
                 Integer pipelineId = jobMetrics.getPipelineId();
                 JobMetrics currentPiplinejobMetricsFromEngine =