Fix the bug when update workflow in DolphinScheduler will cause cycle dependencies.

Author: wushengyeyouya

dss-appconn/appconns/dss-dolphinscheduler-appconn/src/main/java/com/webank/wedatasphere/dss/appconn/dolphinscheduler/conversion/WorkflowToDolphinSchedulerRelConverter.java

@@ -11,7 +11,6 @@
 import com.webank.wedatasphere.dss.workflow.core.entity.Workflow;
 import com.webank.wedatasphere.dss.workflow.core.entity.WorkflowNode;
 import com.webank.wedatasphere.dss.workflow.core.entity.WorkflowNodeEdge;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeanUtils;
@@ -59,14 +58,17 @@ private DolphinSchedulerWorkflow convertWorkflow(DolphinSchedulerConvertedRel do
             put("direct", "IN");
         }});
         Map<String, DolphinSchedulerWorkflow.LocationInfo> locations = new HashMap<>();
+        Map<String, String> nameToIdMap = workflow.getWorkflowNodes().stream()
+                .collect(Collectors.toMap(WorkflowNode::getName, WorkflowNode::getId));
         for (WorkflowNode workflowNode : workflow.getWorkflowNodes()) {
             DSSNode node = workflowNode.getDSSNode();
             DolphinSchedulerTask dolphinSchedulerTask = nodeConverter.convertNode(dolphinSchedulerConvertedRel, node);
             processDefinitionJson.addTask(dolphinSchedulerTask);
 
             DolphinSchedulerWorkflow.LocationInfo locationInfo = new DolphinSchedulerWorkflow.LocationInfo();
             locationInfo.setName(node.getName());
-            locationInfo.setTargetarr(StringUtils.join(node.getDependencys(), ","));
+            String targetarr = node.getDependencys().stream().map(nameToIdMap::get).collect(Collectors.joining(","));
+            locationInfo.setTargetarr(targetarr);
             locationInfo.setX((int)node.getLayout().getX());
             locationInfo.setY((int)node.getLayout().getY());
             locations.put(node.getId(), locationInfo);