third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: missing node status info in workflow instance DAG

Browse Source
This commit is contained in:
Echo009 2021-03-16 19:35:49 +08:00
parent 145d525b29
commit d29f0651e2
1 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/workflow/WorkflowInstanceManager.java
View File

@ -87,7 +87,7 @@ public class WorkflowInstanceManager {
// 构造实例信息 // 构造实例信息
WorkflowInstanceInfoDO newWfInstance = constructWfInstance(wfInfo, initParams, expectTriggerTime, wfId, wfInstanceId); WorkflowInstanceInfoDO newWfInstance = constructWfInstance(wfInfo, initParams, expectTriggerTime, wfId, wfInstanceId);
PEWorkflowDAG dag; PEWorkflowDAG dag = null;
try { try {
dag = JSON.parseObject(wfInfo.getPeDAG(), PEWorkflowDAG.class); dag = JSON.parseObject(wfInfo.getPeDAG(), PEWorkflowDAG.class);
// 校验 DAG 信息 // 校验 DAG 信息
@ -97,7 +97,6 @@ public class WorkflowInstanceManager {
} }
// 初始化节点信息 // 初始化节点信息
initNodeInfo(dag); initNodeInfo(dag);
newWfInstance.setDag(JSON.toJSONString(dag));
// 最后检查工作流中的任务是否均处于可用状态（没有被删除） // 最后检查工作流中的任务是否均处于可用状态（没有被删除）
Set<Long> allJobIds = Sets.newHashSet(); Set<Long> allJobIds = Sets.newHashSet();
dag.getNodes().forEach(node -> { dag.getNodes().forEach(node -> {
@ -112,8 +111,12 @@ public class WorkflowInstanceManager {
log.warn("[Workflow-{}|{}] this workflow need {} jobs, but just find {} jobs in database, maybe you delete or disable some job!", wfId, wfInstanceId, needNum, dbNum); log.warn("[Workflow-{}|{}] this workflow need {} jobs, but just find {} jobs in database, maybe you delete or disable some job!", wfId, wfInstanceId, needNum, dbNum);
throw new PowerJobException(SystemInstanceResult.CAN_NOT_FIND_JOB); throw new PowerJobException(SystemInstanceResult.CAN_NOT_FIND_JOB);
} }
newWfInstance.setDag(JSON.toJSONString(dag));
workflowInstanceInfoRepository.saveAndFlush(newWfInstance); workflowInstanceInfoRepository.saveAndFlush(newWfInstance);
} catch (Exception e) { } catch (Exception e) {
if (dag != null) {
newWfInstance.setDag(JSON.toJSONString(dag));
}
onWorkflowInstanceFailed(e.getMessage(), newWfInstance); onWorkflowInstanceFailed(e.getMessage(), newWfInstance);
} }
return wfInstanceId; return wfInstanceId;