[Feature][JsonSplit-api] update method of ProcessDefinition (#6089)

* refactor method of task save

* fix ut

* fix ut

* update method of processDefinition

* fix ut

Co-authored-by: JinyLeeChina <297062848@qq.com>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java

@@ -175,7 +175,7 @@ public class ExecutorController extends BaseController {
     @ResponseStatus(HttpStatus.OK)
     @ApiException(CHECK_PROCESS_DEFINITION_ERROR)
     @AccessLogAnnotation(ignoreRequestArgs = "loginUser")
-    public Result startCheckProcessDefinition(@RequestParam(value = "processDefinitionCode") int processDefinitionCode) {
+    public Result startCheckProcessDefinition(@RequestParam(value = "processDefinitionCode") long processDefinitionCode) {
         Map<String, Object> result = execService.startCheckByProcessDefinedCode(processDefinitionCode);
         return returnDataList(result);
     }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java

@@ -142,6 +142,18 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
     @Autowired
     private ProcessTaskRelationMapper processTaskRelationMapper;
 
+    @Autowired
+    TaskDefinitionLogMapper taskDefinitionLogMapper;
+
+    @Autowired
+    private TaskDefinitionMapper taskDefinitionMapper;
+
+    @Autowired
+    private SchedulerService schedulerService;
+
+    @Autowired
+    private TenantMapper tenantMapper;
+
     /**
      * create process definition
      *
@@ -193,10 +205,14 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
             return checkRelationJson;
         }
 
-        Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
+        int tenantId = -1;
-        if (tenant == null) {
+        if (!Constants.DEFAULT.equals(tenantCode)) {
-            putMsg(result, Status.TENANT_NOT_EXIST);
+            Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
-            return result;
+            if (tenant == null) {
+                putMsg(result, Status.TENANT_NOT_EXIST);
+                return result;
+            }
+            tenantId = tenant.getId();
         }
 
         long processDefinitionCode;
@@ -207,23 +223,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
             return result;
         }
         ProcessDefinition processDefinition = new ProcessDefinition(projectCode, name, processDefinitionCode, description,
-            globalParams, locations, timeout, loginUser.getId(), tenant.getId());
+            globalParams, locations, timeout, loginUser.getId(), tenantId);
 
         return createProcessDefine(loginUser, result, taskRelationList, processDefinition, taskDefinitionLogs);
     }
 
-    @Autowired
-    TaskDefinitionLogMapper taskDefinitionLogMapper;
-
-    @Autowired
-    private TaskDefinitionMapper taskDefinitionMapper;
-
-    @Autowired
-    private SchedulerService schedulerService;
-
-    @Autowired
-    private TenantMapper tenantMapper;
-
     private void createTaskDefinition(Map<String, Object> result,
                                       User loginUser,
                                       long projectCode,
@@ -240,12 +244,6 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
                 putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskDefinitionLog.getName());
                 return;
             }
-            TaskDefinition taskDefinition = taskDefinitionMapper.queryByName(projectCode, taskDefinitionLog.getName());
-            if (taskDefinition != null) {
-                logger.error("task definition name {} already exists", taskDefinitionLog.getName());
-                putMsg(result, Status.TASK_DEFINITION_NAME_EXISTED, taskDefinitionLog.getName());
-                return;
-            }
         }
         if (processService.saveTaskDefine(loginUser, projectCode, taskDefinitionLogs)) {
             putMsg(result, Status.SUCCESS);
@@ -401,6 +399,10 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
         if (processDefinition == null) {
             putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, code);
         } else {
+            Tenant tenant = tenantMapper.queryById(processDefinition.getTenantId());
+            if (tenant != null) {
+                processDefinition.setTenantCode(tenant.getTenantCode());
+            }
             DagData dagData = processService.genDagData(processDefinition);
             result.put(Constants.DATA_LIST, dagData);
             putMsg(result, Status.SUCCESS);
@@ -475,10 +477,14 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
             return checkRelationJson;
         }
 
-        Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
+        int tenantId = -1;
-        if (tenant == null) {
+        if (!Constants.DEFAULT.equals(tenantCode)) {
-            putMsg(result, Status.TENANT_NOT_EXIST);
+            Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
-            return result;
+            if (tenant == null) {
+                putMsg(result, Status.TENANT_NOT_EXIST);
+                return result;
+            }
+            tenantId = tenant.getId();
         }
 
         ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(code);
@@ -500,21 +506,28 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
                 return result;
             }
         }
-        processDefinition.set(projectCode, name, description, globalParams, locations, timeout, tenant.getId());
+        ProcessDefinition processDefinitionDeepCopy = JSONUtils.parseObject(JSONUtils.toJsonString(processDefinition), ProcessDefinition.class);
-        return updateProcessDefine(loginUser, result, taskRelationList, processDefinition, taskDefinitionLogs);
+        processDefinition.set(projectCode, name, description, globalParams, locations, timeout, tenantId);
+        return updateProcessDefine(loginUser, result, taskRelationList, processDefinition, processDefinitionDeepCopy, taskDefinitionLogs);
     }
 
     private Map<String, Object> updateProcessDefine(User loginUser,
                                                     Map<String, Object> result,
                                                     List<ProcessTaskRelationLog> taskRelationList,
                                                     ProcessDefinition processDefinition,
+                                                    ProcessDefinition processDefinitionDeepCopy,
                                                     List<TaskDefinitionLog> taskDefinitionLogs) {
-        processDefinition.setUpdateTime(new Date());
+        int insertVersion;
-        int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, true);
+        if (processDefinition.equals(processDefinitionDeepCopy)) {
+            insertVersion = processDefinitionDeepCopy.getVersion();
+        } else {
+            processDefinition.setUpdateTime(new Date());
+            insertVersion = processService.saveProcessDefine(loginUser, processDefinition, true);
+        }
         if (insertVersion > 0) {
             int insertResult = processService.saveTaskRelation(loginUser, processDefinition.getProjectCode(),
                 processDefinition.getCode(), insertVersion, taskRelationList, taskDefinitionLogs);
-            if (insertResult > 0) {
+            if (insertResult == Constants.EXIT_CODE_SUCCESS) {
                 putMsg(result, Status.SUCCESS);
                 result.put(Constants.DATA_LIST, processDefinition);
             } else {
@@ -1286,7 +1299,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
                 processDefinition.setName(processDefinition.getName() + "_copy_" + DateUtils.getCurrentTimeStamp());
                 createProcessDefine(loginUser, result, taskRelationList, processDefinition, Lists.newArrayList());
             } else {
-                updateProcessDefine(loginUser, result, taskRelationList, processDefinition, Lists.newArrayList());
+                updateProcessDefine(loginUser, result, taskRelationList, processDefinition, null, Lists.newArrayList());
             }
             if (result.get(Constants.STATUS) != Status.SUCCESS) {
                 failedProcessList.add(processDefinition.getCode() + "[" + processDefinition.getName() + "]");

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java

@@ -459,13 +459,17 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce
             if (result.get(Constants.STATUS) != Status.SUCCESS) {
                 return result;
             }
-            Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
+            int tenantId = -1;
-            if (tenant == null) {
+            if (!Constants.DEFAULT.equals(tenantCode)) {
-                putMsg(result, Status.TENANT_NOT_EXIST);
+                Tenant tenant = tenantMapper.queryByTenantCode(tenantCode);
-                return result;
+                if (tenant == null) {
+                    putMsg(result, Status.TENANT_NOT_EXIST);
+                    return result;
+                }
+                tenantId = tenant.getId();
             }
 
-            processDefinition.set(projectCode, processDefinition.getName(), processDefinition.getDescription(), globalParams, locations, timeout, tenant.getId());
+            processDefinition.set(projectCode, processDefinition.getName(), processDefinition.getDescription(), globalParams, locations, timeout, tenantId);
             processDefinition.setUpdateTime(new Date());
             int insertVersion = processService.saveProcessDefine(loginUser, processDefinition, false);
             if (insertVersion > 0) {

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java

@@ -36,12 +36,14 @@ import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
 import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
 import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.Schedule;
+import org.apache.dolphinscheduler.dao.entity.Tenant;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
+import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 
 import java.text.MessageFormat;
@@ -96,6 +98,8 @@ public class ProcessDefinitionServiceTest {
     private ProcessInstanceService processInstanceService;
     @Mock
     private TaskInstanceMapper taskInstanceMapper;
+    @Mock
+    private TenantMapper tenantMapper;
 
     @Test
     public void testQueryProcessDefinitionList() {
@@ -173,7 +177,9 @@ public class ProcessDefinitionServiceTest {
         User loginUser = new User();
         loginUser.setId(-1);
         loginUser.setUserType(UserType.GENERAL_USER);
-
+        Tenant tenant = new Tenant();
+        tenant.setId(1);
+        tenant.setTenantCode("root");
         Map<String, Object> result = new HashMap<>();
         putMsg(result, Status.PROJECT_NOT_FOUNT, projectCode);
 
@@ -195,6 +201,7 @@ public class ProcessDefinitionServiceTest {
         Mockito.when(processDefineMapper.queryByCode(46L)).thenReturn(getProcessDefinition());
         putMsg(result, Status.SUCCESS, projectCode);
         Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectCode)).thenReturn(result);
+        Mockito.when(tenantMapper.queryById(1)).thenReturn(tenant);
         Map<String, Object> successRes = processDefinitionService.queryProcessDefinitionByCode(loginUser, projectCode, 46L);
         Assert.assertEquals(Status.SUCCESS, successRes.get(Constants.STATUS));
     }

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java

@@ -435,7 +435,7 @@ public final class Constants {
      */
     public static final String DATASOURCE_PROPERTIES = "/datasource.properties";
 
-    public static final String DEFAULT = "Default";
+    public static final String DEFAULT = "default";
     public static final String USER = "user";
     public static final String PASSWORD = "password";
     public static final String XXXXXX = "******";

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/TaskTimeoutStrategy.java

@@ -28,11 +28,10 @@ public enum TaskTimeoutStrategy {
      * 2 warn+failed
      */
     WARN(0, "warn"),
-    FAILED(1,"failed"),
+    FAILED(1, "failed"),
-    WARNFAILED(2,"warnfailed");
+    WARNFAILED(2, "warnfailed");
 
-
+    TaskTimeoutStrategy(int code, String descp) {
-    TaskTimeoutStrategy(int code, String descp){
         this.code = code;
         this.descp = descp;
     }
@@ -49,9 +48,9 @@ public enum TaskTimeoutStrategy {
         return descp;
     }
 
-    public static TaskTimeoutStrategy of(int status){
+    public static TaskTimeoutStrategy of(int status) {
-        for(TaskTimeoutStrategy es : values()){
+        for (TaskTimeoutStrategy es : values()) {
-            if(es.getCode() == status){
+            if (es.getCode() == status) {
                 return es;
             }
         }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinition.java

@@ -26,6 +26,7 @@ import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 import com.baomidou.mybatisplus.annotation.IdType;
@@ -149,6 +150,12 @@ public class ProcessDefinition {
      */
     private int tenantId;
 
+    /**
+     * tenant code
+     */
+    @TableField(exist = false)
+    private String tenantCode;
+
     /**
      * modify user name
      */
@@ -167,7 +174,8 @@ public class ProcessDefinition {
     @TableField(exist = false)
     private int warningGroupId;
 
-    public ProcessDefinition() {}
+    public ProcessDefinition() {
+    }
 
     public ProcessDefinition(long projectCode,
                              String name,
@@ -356,6 +364,14 @@ public class ProcessDefinition {
         this.tenantId = tenantId;
     }
 
+    public String getTenantCode() {
+        return tenantCode;
+    }
+
+    public void setTenantCode(String tenantCode) {
+        this.tenantCode = tenantCode;
+    }
+
     public String getDescription() {
         return description;
     }
@@ -396,12 +412,33 @@ public class ProcessDefinition {
         this.warningGroupId = warningGroupId;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ProcessDefinition that = (ProcessDefinition) o;
+        return projectCode == that.projectCode
+            && userId == that.userId
+            && timeout == that.timeout
+            && tenantId == that.tenantId
+            && Objects.equals(name, that.name)
+            && releaseState == that.releaseState
+            && Objects.equals(description, that.description)
+            && Objects.equals(globalParams, that.globalParams)
+            && flag == that.flag
+            && Objects.equals(locations, that.locations);
+    }
+
     @Override
     public String toString() {
         return "ProcessDefinition{"
             + "id=" + id
-            + ", name='" + name + '\''
             + ", code=" + code
+            + ", name='" + name + '\''
             + ", version=" + version
             + ", releaseState=" + releaseState
             + ", projectCode=" + projectCode
@@ -418,10 +455,11 @@ public class ProcessDefinition {
             + ", locations='" + locations + '\''
             + ", scheduleReleaseState=" + scheduleReleaseState
             + ", timeout=" + timeout
-            + ", warningGroupId=" + warningGroupId
             + ", tenantId=" + tenantId
+            + ", tenantCode='" + tenantCode + '\''
             + ", modifyBy='" + modifyBy + '\''
             + ", resourceIds='" + resourceIds + '\''
+            + ", warningGroupId=" + warningGroupId
             + '}';
     }
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionLog.java

@@ -85,4 +85,8 @@ public class ProcessDefinitionLog extends ProcessDefinition {
         this.operateTime = operateTime;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        return super.equals(o);
+    }
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessTaskRelation.java

@@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.enums.ConditionType;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 
 import java.util.Date;
+import java.util.Objects;
 
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableId;
@@ -238,6 +239,30 @@ public class ProcessTaskRelation {
         this.postTaskVersion = postTaskVersion;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ProcessTaskRelation that = (ProcessTaskRelation) o;
+        return processDefinitionVersion == that.processDefinitionVersion
+            && projectCode == that.projectCode
+            && processDefinitionCode == that.processDefinitionCode
+            && preTaskCode == that.preTaskCode
+            && preTaskVersion == that.preTaskVersion
+            && postTaskCode == that.postTaskCode
+            && postTaskVersion == that.postTaskVersion
+            && Objects.equals(name, that.name);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name, processDefinitionVersion, projectCode, processDefinitionCode, preTaskCode, preTaskVersion, postTaskCode, postTaskVersion);
+    }
+
     @Override
     public String toString() {
         return "ProcessTaskRelation{"

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessTaskRelationLog.java

@@ -76,6 +76,16 @@ public class ProcessTaskRelationLog extends ProcessTaskRelation {
         this.operateTime = operateTime;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        return super.equals(o);
+    }
+
+    @Override
+    public int hashCode() {
+        return super.hashCode();
+    }
+
     @Override
     public String toString() {
         return super.toString();

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java

@@ -2135,8 +2135,9 @@ public class ProcessService {
             if (task == null) {
                 newTaskDefinitionLogs.add(taskDefinitionToUpdate);
             } else {
-                int update = taskDefinitionMapper.updateById(taskDefinitionToUpdate);
                 int insert = taskDefinitionLogMapper.insert(taskDefinitionToUpdate);
+                taskDefinitionToUpdate.setId(task.getId());
+                int update = taskDefinitionMapper.updateById(taskDefinitionToUpdate);
                 if ((update & insert) != 1) {
                     return false;
                 }
@@ -2177,10 +2178,6 @@ public class ProcessService {
      */
     public int saveTaskRelation(User operator, long projectCode, long processDefinitionCode, int processDefinitionVersion,
                                 List<ProcessTaskRelationLog> taskRelationList, List<TaskDefinitionLog> taskDefinitionLogs) {
-        List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode);
-        if (!processTaskRelationList.isEmpty()) {
-            processTaskRelationMapper.deleteByCode(projectCode, processDefinitionCode);
-        }
         Map<Long, TaskDefinitionLog> taskDefinitionLogMap = null;
         if (CollectionUtils.isNotEmpty(taskDefinitionLogs)) {
             taskDefinitionLogMap = taskDefinitionLogs.stream()
@@ -2203,12 +2200,21 @@ public class ProcessService {
             processTaskRelationLog.setOperator(operator.getId());
             processTaskRelationLog.setOperateTime(now);
         }
+        List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByProcessCode(projectCode, processDefinitionCode);
+        if (!processTaskRelationList.isEmpty()) {
+            Set<Integer> processTaskRelationSet = processTaskRelationList.stream().map(ProcessTaskRelation::hashCode).collect(toSet());
+            Set<Integer> taskRelationSet = taskRelationList.stream().map(ProcessTaskRelationLog::hashCode).collect(toSet());
+            if (CollectionUtils.isEqualCollection(processTaskRelationSet, taskRelationSet)) {
+                return Constants.EXIT_CODE_SUCCESS;
+            }
+            processTaskRelationMapper.deleteByCode(projectCode, processDefinitionCode);
+        }
         int result = processTaskRelationMapper.batchInsert(taskRelationList);
         int resultLog = processTaskRelationLogMapper.batchInsert(taskRelationList);
-        return result & resultLog;
+        return (result & resultLog) > 0 ? Constants.EXIT_CODE_SUCCESS : Constants.EXIT_CODE_FAILURE;
     }
 
-    public boolean isTaskOnline(Long taskCode) {
+    public boolean isTaskOnline(long taskCode) {
         List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByTaskCode(taskCode);
         if (!processTaskRelationList.isEmpty()) {
             Set<Long> processDefinitionCodes = processTaskRelationList