dss-server:fix code

WeDataSphere · Oct 18, 2024 · 861ebef · 861ebef
1 parent eaa421e
commit 861ebef
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 1 deletion.
diff --git a/...nk/wedatasphere/dss/flow/execution/entrance/conf/FlowExecutionEntranceConfiguration.scala b/...nk/wedatasphere/dss/flow/execution/entrance/conf/FlowExecutionEntranceConfiguration.scala
@@ -77,4 +77,10 @@ object FlowExecutionEntranceConfiguration {
 
   val CONFIGURATION = "configuration"
 
+  val COMMAND = "command"
+
+  val USER_TO_PROXY = "user.to.proxy"
+
+  val FLOW_VAR_MAP = "flow_var_map"
+
 }
diff --git a/.../com/webank/wedatasphere/dss/flow/execution/entrance/execution/DefaultFlowExecution.scala b/.../com/webank/wedatasphere/dss/flow/execution/entrance/execution/DefaultFlowExecution.scala
@@ -74,7 +74,15 @@ class DefaultFlowExecution extends FlowExecution with Logging {
         info(s"${flowEntranceJob.getId} Submit nodes(${runningNodes.size}) to running")
         runningNodes.foreach { node =>
           node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.PROPS_MAP).asInstanceOf[java.util.Map[String, Any]].putAll(flowEntranceJob.getParams)
-          node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.CONFIGURATION).asInstanceOf[java.util.Map[String, Any]].get(FlowExecutionEntranceConfiguration.START_UP).asInstanceOf[java.util.Map[String, Any]].put(FlowExecutionEntranceConfiguration.YARN_QUEUE, null)
+          if (node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.CONFIGURATION) != null && node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.CONFIGURATION).asInstanceOf[java.util.Map[String, Any]].get(FlowExecutionEntranceConfiguration.START_UP) != null) {
+            node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.CONFIGURATION).asInstanceOf[java.util.Map[String, Any]].get(FlowExecutionEntranceConfiguration.START_UP).asInstanceOf[java.util.Map[String, Any]].put(FlowExecutionEntranceConfiguration.YARN_QUEUE, null)
+          }
+          if (node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.PROPS_MAP) != null && node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.PROPS_MAP).asInstanceOf[java.util.Map[String, Any]].get(FlowExecutionEntranceConfiguration.COMMAND) != null) {
+            node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.PROPS_MAP).asInstanceOf[java.util.Map[String, Any]].get(FlowExecutionEntranceConfiguration.COMMAND).asInstanceOf[java.util.Map[String, Any]].put(FlowExecutionEntranceConfiguration.USER_TO_PROXY, null)
+          }
+          if (node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.FLOW_VAR_MAP) != null) {
+            node.getNode.getDSSNode.getParams.get(FlowExecutionEntranceConfiguration.FLOW_VAR_MAP).asInstanceOf[java.util.Map[String, Any]].put(FlowExecutionEntranceConfiguration.USER_TO_PROXY, null)
+          }
           info(s"nodeParams:${node.getNode.getDSSNode.getParams} ")
           info(s"JobContent:${node.getNode.getDSSNode.getJobContent} ")
           node.run()