[Improvement][API][num-1] save schedule when workflow is offline in interface 'createEmptyProcessDefinition' (#7047)

* [Improvement] save schedule when workflow is offline

* [Improvement] rollback transactional when creating/updating schedule failed.

* [Improvement] resolve merge conflict

* [Improvement] do not set projectName and processDefinitionName when saving schedule

* [Improvement] do not set projectName and processDefinitionName when saving schedule

Co-authored-by: edward-yang <yangjianh210@gmail.com>

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

@@ -1596,9 +1596,11 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
         }
 
         // save dag schedule
-        Map<String, Object> scheduleResult = createDagSchedule(loginUser, projectCode, processDefinitionCode, scheduleJson);
+        Map<String, Object> scheduleResult = createDagSchedule(loginUser, project, processDefinition, scheduleJson);
         if (scheduleResult.get(Constants.STATUS) != Status.SUCCESS) {
-            return scheduleResult;
+            Status scheduleResultStatus = (Status) scheduleResult.get(Constants.STATUS);
+            putMsg(result, scheduleResultStatus);
+            throw new ServiceException(scheduleResultStatus);
         }
         return result;
     }
@@ -1616,40 +1618,48 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
     }
 
     private Map<String, Object> createDagSchedule(User loginUser,
-                                                  long projectCode,
+                                                  Project project,
-                                                  long processDefinitionCode,
+                                                  ProcessDefinition processDefinition,
                                                   String scheduleJson) {
         Map<String, Object> result = new HashMap<>();
-        Schedule schedule = JSONUtils.parseObject(scheduleJson, Schedule.class);
+        Schedule scheduleObj = JSONUtils.parseObject(scheduleJson, Schedule.class);
-        if (schedule == null) {
+        if (scheduleObj == null) {
             putMsg(result, Status.DATA_IS_NOT_VALID, scheduleJson);
             throw new ServiceException(Status.DATA_IS_NOT_VALID);
         }
-        // set default value
+        Date now = new Date();
-        FailureStrategy failureStrategy = schedule.getFailureStrategy() == null ? FailureStrategy.CONTINUE : schedule.getFailureStrategy();
+        scheduleObj.setProcessDefinitionCode(processDefinition.getCode());
-        WarningType warningType = schedule.getWarningType() == null ? WarningType.NONE : schedule.getWarningType();
+        if (DateUtils.differSec(scheduleObj.getStartTime(), scheduleObj.getEndTime()) == 0) {
-        Priority processInstancePriority = schedule.getProcessInstancePriority() == null ? Priority.MEDIUM : schedule.getProcessInstancePriority();
+            logger.warn("The start time must not be the same as the end");
-        int warningGroupId = schedule.getWarningGroupId() == 0 ? 1 : schedule.getWarningGroupId();
+            putMsg(result, Status.SCHEDULE_START_TIME_END_TIME_SAME);
-        String workerGroup = schedule.getWorkerGroup() == null ? "default" : schedule.getWorkerGroup();
+            return result;
-        long environmentCode = schedule.getEnvironmentCode() == null ? -1 : schedule.getEnvironmentCode();
+        }
-
+        if (!org.quartz.CronExpression.isValidExpression(scheduleObj.getCrontab())) {
-        ScheduleParam param = new ScheduleParam();
+            logger.error("{} verify failure", scheduleObj.getCrontab());
-        param.setStartTime(schedule.getStartTime());
+            putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, scheduleObj.getCrontab());
-        param.setEndTime(schedule.getEndTime());
+            return result;
-        param.setCrontab(schedule.getCrontab());
+        }
-        param.setTimezoneId(schedule.getTimezoneId());
+        scheduleObj.setWarningType(scheduleObj.getWarningType() == null ? WarningType.NONE : scheduleObj.getWarningType());
+        scheduleObj.setWarningGroupId(scheduleObj.getWarningGroupId() == 0 ? 1 : scheduleObj.getWarningGroupId());
+        scheduleObj.setFailureStrategy(scheduleObj.getFailureStrategy() == null ? FailureStrategy.CONTINUE : scheduleObj.getFailureStrategy());
+        scheduleObj.setCreateTime(now);
+        scheduleObj.setUpdateTime(now);
+        scheduleObj.setUserId(loginUser.getId());
+        scheduleObj.setReleaseState(ReleaseState.OFFLINE);
+        scheduleObj.setProcessInstancePriority(scheduleObj.getProcessInstancePriority() == null ? Priority.MEDIUM : scheduleObj.getProcessInstancePriority());
+        scheduleObj.setWorkerGroup(scheduleObj.getWorkerGroup() == null ? "default" : scheduleObj.getWorkerGroup());
+        scheduleObj.setEnvironmentCode(scheduleObj.getEnvironmentCode() == null ? -1 : scheduleObj.getEnvironmentCode());
+        scheduleMapper.insert(scheduleObj);
+
+        /**
+         * updateProcessInstance receivers and cc by process definition id
+         */
+        processDefinition.setWarningGroupId(scheduleObj.getWarningGroupId());
+        processDefinitionMapper.updateById(processDefinition);
 
-        return schedulerService.insertSchedule(
+        putMsg(result, Status.SUCCESS);
-                loginUser,
+        result.put("scheduleId", scheduleObj.getId());
-                projectCode,
+        return result;
-                processDefinitionCode,
-                JSONUtils.toJsonString(param),
-                warningType,
-                warningGroupId,
-                failureStrategy,
-                processInstancePriority,
-                workerGroup,
-                environmentCode);
     }
 
     /**
@@ -1668,6 +1678,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
      * @return update result code
      */
     @Override
+    @Transactional(rollbackFor = RuntimeException.class)
     public Map<String, Object> updateProcessDefinitionBasicInfo(User loginUser,
                                                                 long projectCode,
                                                                 String name,
@@ -1728,7 +1739,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
         // update dag schedule
         Map<String, Object> scheduleResult = updateDagSchedule(loginUser, projectCode, code, scheduleJson);
         if (scheduleResult.get(Constants.STATUS) != Status.SUCCESS) {
-            return scheduleResult;
+            Status scheduleResultStatus = (Status) scheduleResult.get(Constants.STATUS);
+            putMsg(result, scheduleResultStatus);
+            throw new ServiceException(scheduleResultStatus);
         }
         return result;
     }