fix dag empty msg (#5245)

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java

@@ -262,6 +262,7 @@ public enum Status {
     EXPORT_PROCESS_DEFINE_BY_ID_ERROR(50028, "export process definition by id error", "导出工作流定义错误"),
     BATCH_EXPORT_PROCESS_DEFINE_BY_IDS_ERROR(50028, "batch export process definition by ids error", "批量导出工作流定义错误"),
     IMPORT_PROCESS_DEFINE_ERROR(50029, "import process definition error", "导入工作流定义错误"),
+    PROCESS_DAG_IS_EMPTY(50030, "process dag can not be empty", "工作流dag不能为空"),
 
     HDFS_NOT_STARTUP(60001, "hdfs not startup", "hdfs未启用"),
 

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

@@ -1168,9 +1168,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
             // Check whether the task node is normal
             List<TaskNode> taskNodes = processData.getTasks();
 
-            if (taskNodes == null) {
+            if (CollectionUtils.isEmpty(taskNodes)) {
                 logger.error("process node info is empty");
-                putMsg(result, Status.DATA_IS_NULL, processDefinitionJson);
+                putMsg(result, Status.PROCESS_DAG_IS_EMPTY);
                 return result;
             }
 

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

@@ -679,7 +679,7 @@ public class ProcessDefinitionServiceTest {
         // task empty
         processData.setTasks(null);
         Map<String, Object> taskNotEmptyRes = processDefinitionService.checkProcessNodeList(processData, processDefinitionJson);
-        Assert.assertEquals(Status.DATA_IS_NULL, taskNotEmptyRes.get(Constants.STATUS));
+        Assert.assertEquals(Status.PROCESS_DAG_IS_EMPTY, taskNotEmptyRes.get(Constants.STATUS));
 
         // task cycle
         String processDefinitionJsonCycle = CYCLE_SHELL_JSON;