Merge pull request #998 from WeDataSphere/master

Fix some bug abut intergrating DolphinScheduler.

Author: zqburde

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

@@ -5,13 +5,13 @@
 import com.webank.wedatasphere.dss.appconn.dolphinscheduler.entity.DolphinSchedulerWorkflow;
 import com.webank.wedatasphere.dss.common.entity.node.DSSNode;
 import com.webank.wedatasphere.dss.common.exception.DSSRuntimeException;
+import com.webank.wedatasphere.dss.standard.common.exception.operation.ExternalOperationFailedException;
 import com.webank.wedatasphere.dss.workflow.conversion.entity.ConvertedRel;
 import com.webank.wedatasphere.dss.workflow.conversion.entity.PreConversionRel;
 import com.webank.wedatasphere.dss.workflow.conversion.operation.WorkflowToRelConverter;
 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;
@@ -31,6 +31,12 @@ public class WorkflowToDolphinSchedulerRelConverter implements WorkflowToRelConv
     @Override
     public ConvertedRel convertToRel(PreConversionRel rel) {
         DolphinSchedulerConvertedRel dolphinSchedulerConvertedRel = (DolphinSchedulerConvertedRel)rel;
+        List<WorkflowNode> workflowNodes = dolphinSchedulerConvertedRel.getWorkflow().getWorkflowNodes();
+        for (WorkflowNode workflowNode : workflowNodes) {
+            if ("workflow.subflow".equals(workflowNode.getNodeType())) {
+                throw new ExternalOperationFailedException(90021, workflowNode.getName() + "当前不支持将subFlow节点发布到DolphinScheduler！");
+            }
+        }
         Workflow dolphinSchedulerWorkflow = convertWorkflow(dolphinSchedulerConvertedRel);
         dolphinSchedulerConvertedRel.setWorkflow(dolphinSchedulerWorkflow);
         return dolphinSchedulerConvertedRel;
@@ -59,14 +65,20 @@ 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();
+            if (node.getLayout() == null) {
+                continue;
+            }
             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);

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

@@ -36,7 +36,7 @@ public void syncToRel(ConvertedRel convertedRel) {
         DolphinSchedulerConvertedRel dolphinSchedulerConvertedRel = (DolphinSchedulerConvertedRel)convertedRel;
         OrchestrationToRelConversionRequestRef requestRef = dolphinSchedulerConvertedRel.getDSSToRelConversionRequestRef();
         Workflow workflow = dolphinSchedulerConvertedRel.getWorkflow();
-        checkSchedulerProject(workflow);
+//        checkSchedulerProject(workflow);
         Long dolphinSchedulerWorkflowId = requestRef.getRefOrchestrationId();
         DolphinSchedulerAppConn appConn = (DolphinSchedulerAppConn) dssToRelConversionOperation.getConversionService().getAppStandard().getAppConn();
         OrchestrationUpdateOperation updateOperation = appConn.getOrCreateStructureStandard().getOrchestrationService(dssToRelConversionOperation.getConversionService().getAppInstance())
@@ -47,14 +47,14 @@ public void syncToRel(ConvertedRel convertedRel) {
         updateOperation.updateOrchestration(ref);
     }
 
-    private void checkSchedulerProject(Workflow flow) throws ExternalOperationFailedException {
-        List<WorkflowNode> nodes = flow.getWorkflowNodes();
-        for (WorkflowNode node : nodes) {
-            DSSNode dssNode = node.getDSSNode();
-            if (CollectionUtils.isEmpty(dssNode.getResources()) && dssNode.getJobContent().isEmpty()) {
-                throw new ExternalOperationFailedException(90021, dssNode.getName() + "节点内容不能为空");
-            }
-        }
-    }
+//    private void checkSchedulerProject(Workflow flow) throws ExternalOperationFailedException {
+//        List<WorkflowNode> nodes = flow.getWorkflowNodes();
+//        for (WorkflowNode node : nodes) {
+//            DSSNode dssNode = node.getDSSNode();
+//            if (CollectionUtils.isEmpty(dssNode.getResources()) && dssNode.getJobContent().isEmpty()) {
+//                throw new ExternalOperationFailedException(90021, dssNode.getName() + "节点内容不能为空");
+//            }
+//        }
+//    }
 
 }

dss-appconn/appconns/dss-dolphinscheduler-appconn/src/main/scala/com/webank/wedatasphere/dss/appconn/schedulis/conf/SchedulisConf.scala

@@ -55,7 +55,7 @@ public Job build() throws Exception {
         LinkisJob job = null;
         String jobType = getJobType();
         String[] jobTypeSplit = jobType.split("\\.");
-        if (jobTypeSplit.length < 3) {
+        if (jobTypeSplit.length < 2) {
             throw new LinkisJobExecutionErrorException(90100, "This is not Linkis job type, this jobtype is " + jobType);
         }
         String engineType = jobTypeSplit[1];