fix: 适配调度系统#AI Commit#

Author: zqburde

plugins/azkaban/linkis-jobtype/src/main/java/com/webank/wedatasphere/dss/plugins/azkaban/linkis/jobtype/AzkabanDssJobType.java

@@ -61,6 +61,8 @@ public class AzkabanDssJobType extends AbstractJob {
 
     private Job job;
 
+    private volatile Props generatedProperties = new Props();
+
     private boolean isCanceled = false;
 
     public AzkabanDssJobType(String jobId, Props sysProps, Props jobProps, Logger log) {
@@ -151,6 +153,11 @@ public boolean isCanceled() {
         return isCanceled;
     }
 
+    @Override
+    public Props getJobGeneratedProperties() {
+        return this.generatedProperties == null ? new Props() : this.generatedProperties;
+    }
+
     @Override
     public double getProgress() throws Exception {
         return LinkisNodeExecutionImpl.getLinkisNodeExecution().getProgress(this.job);
@@ -165,6 +172,9 @@ private void collectBranchVariables() {
             Map<String, String> resultVariables = LinkisNodeExecutionImpl.getLinkisNodeExecution().getResultVariables(this.job, 128);
             Map<String, String> resolvedVariables = resolveBranchOutputVariables(resultVariables);
             if (!resolvedVariables.isEmpty()) {
+                Props props = new Props();
+                props.putAll(resolvedVariables);
+                this.generatedProperties = props;
                 BranchRuntimeStore.mergeFlowVariables(flowExecId, resolvedVariables);
                 info("Collected branch flow variables: " + resolvedVariables);
             }
@@ -297,3 +307,4 @@ private String getRunTodayh(boolean stdFormat) {
         return null;
     }
 }
+

plugins/azkaban/linkis-jobtype/src/main/java/com/webank/wedatasphere/dss/plugins/azkaban/linkis/jobtype/job/BranchRouteExecutor.java

@@ -57,18 +57,34 @@ public void execute() throws Exception {
 
     private Map<String, String> buildEvaluationContext(String branchNodeId, String branchNodeName) {
         Map<String, String> context = new LinkedHashMap<>();
+        for (Map.Entry<String, String> entry : jobProps.entrySet()) {
+            if (!isBlank(entry.getKey()) && entry.getValue() != null) {
+                context.put(entry.getKey(), entry.getValue());
+            }
+        }
         for (Map.Entry<String, String> entry : jobProps.entrySet()) {
             if (entry.getKey().startsWith(LinkisJobTypeConf.FLOW_VARIABLE_PREFIX) && entry.getValue() != null) {
                 context.put(entry.getKey().substring(LinkisJobTypeConf.FLOW_VARIABLE_PREFIX.length()), entry.getValue());
             }
         }
-        context.putAll(BranchRuntimeStore.snapshotFlowVariables(jobProps.get(LinkisJobTypeConf.FLOW_EXEC_ID)));
+        mergeMissingVariables(context, BranchRuntimeStore.snapshotFlowVariables(jobProps.get(LinkisJobTypeConf.FLOW_EXEC_ID)));
         context.put("node.id", branchNodeId);
         context.put("node.name", branchNodeName == null ? "" : branchNodeName);
         context.put("node.type", LinkisJobTypeConf.BRANCH_ROUTE_LINKIS_TYPE);
         return context;
     }
 
+    private void mergeMissingVariables(Map<String, String> context, Map<String, String> fallbackVariables) {
+        if (fallbackVariables == null || fallbackVariables.isEmpty()) {
+            return;
+        }
+        for (Map.Entry<String, String> entry : fallbackVariables.entrySet()) {
+            if (!isBlank(entry.getKey()) && entry.getValue() != null && !context.containsKey(entry.getKey())) {
+                context.put(entry.getKey(), entry.getValue());
+            }
+        }
+    }
+
     private String selectTarget(String branchRuleText, Map<String, String> targetNameToId, Map<String, String> context) {
         List<BranchExpressionUtils.BranchRule> rules = BranchExpressionUtils.parseBranchRules(branchRuleText);
         for (BranchExpressionUtils.BranchRule rule : rules) {
@@ -114,3 +130,4 @@ private boolean isBlank(String value) {
         return value == null || value.trim().isEmpty();
     }
 }
+