From 31e79d10ea230da9e9470cd06134e23129b547b7 Mon Sep 17 00:00:00 2001 From: Wenjun Ruan Date: Wed, 5 Jun 2024 23:41:52 +0800 Subject: [PATCH] Fix task nodes under switch task will not be skipped (#16109) (cherry picked from commit f687a7d851c206ae30377ac3f9dc025cec35ffb2) --- .../task/switchtask/SwitchLogicTask.java | 39 +++++++++++++------ .../server/master/utils/SwitchTaskUtils.java | 5 --- .../master/utils/SwitchTaskUtilsTest.java | 12 +----- .../task/api/log/TaskInstanceLogHeader.java | 8 ++-- 4 files changed, 33 insertions(+), 31 deletions(-) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/switchtask/SwitchLogicTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/switchtask/SwitchLogicTask.java index 1f52f9287d..58f2105394 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/switchtask/SwitchLogicTask.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/switchtask/SwitchLogicTask.java @@ -61,26 +61,30 @@ public class SwitchLogicTask extends BaseSyncLogicTask { this.taskInstance = workflowExecuteRunnable.getTaskInstance(taskExecutionContext.getTaskInstanceId()) .orElseThrow(() -> new LogicTaskInitializeException( "Cannot find the task instance in workflow execute runnable")); + // Since the default branch is not in the dependTaskList, we need to add it to the end + // otherwise the default branch will never be skipped in DAGHelper + addDefaultBranchToEnd(); } @Override public void handle() throws MasterTaskExecuteException { - // Calculate the condition result and get the next node if (CollectionUtils.isEmpty(taskParameters.getDependTaskList())) { + // If the branch is empty then will go into the default branch + // This case shouldn't happen, we can directly throw exception and forbid the user to set branch moveToDefaultBranch(); } else { calculateSwitchBranch(); } taskInstance.setSwitchDependency(taskParameters); - log.info("Switch task execute finished"); taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.SUCCESS); + log.info("Switch task execute finished: {}", taskExecutionContext.getCurrentExecutionStatus().name()); } private void moveToDefaultBranch() { - checkIfBranchExist(taskParameters.getNextNode()); - List switchResultVos = taskParameters.getDependTaskList(); - switchResultVos.add(new SwitchResultVo(null, taskParameters.getNextNode())); + SwitchResultVo defaultSwitchResultVo = getDefaultSwitchResultVo(); + checkIfBranchExist(defaultSwitchResultVo.getNextNode()); + taskParameters.setResultConditionLocation(switchResultVos.size() - 1); log.info("The condition is not satisfied, move to the default branch: {}", @@ -90,9 +94,6 @@ public class SwitchLogicTask extends BaseSyncLogicTask { private void calculateSwitchBranch() { List switchResultVos = taskParameters.getDependTaskList(); - if (CollectionUtils.isEmpty(switchResultVos)) { - moveToDefaultBranch(); - } Map globalParams = taskExecutionContext.getPrepareParamsMap(); Map varParams = JSONUtils .toList(taskInstance.getVarPool(), Property.class) @@ -100,7 +101,8 @@ public class SwitchLogicTask extends BaseSyncLogicTask { .collect(Collectors.toMap(Property::getProp, Property -> Property)); int finalConditionLocation = -1; - for (int i = 0; i < switchResultVos.size(); i++) { + // The last one is the default branch, no need to calculate + for (int i = 0; i < switchResultVos.size() - 1; i++) { SwitchResultVo switchResultVo = switchResultVos.get(i); log.info("Begin to execute {} condition: {} ", i, switchResultVo.getCondition()); String content = SwitchTaskUtils.generateContentWithTaskParams(switchResultVo.getCondition(), globalParams, @@ -111,14 +113,18 @@ public class SwitchLogicTask extends BaseSyncLogicTask { result = SwitchTaskUtils.evaluate(content); log.info("Execute condition sentence: {} successfully: {}", content, result); if (result) { + // If matched, break the loop finalConditionLocation = i; + break; } } catch (Exception e) { log.info("Execute condition sentence: {} failed", content, e); } } + // If the finalConditionLocation is -1, then the default branch will be executed if (finalConditionLocation >= 0) { - checkIfBranchExist(switchResultVos.get(finalConditionLocation).getNextNode()); + List nextNodes = switchResultVos.get(finalConditionLocation).getNextNode(); + checkIfBranchExist(nextNodes); log.info("The condition is satisfied, move to the branch: {}", switchResultVos.get(finalConditionLocation).getNextNode().stream() .map(node -> workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowGraph().getDag() @@ -126,7 +132,6 @@ public class SwitchLogicTask extends BaseSyncLogicTask { .collect(Collectors.toList())); taskParameters.setResultConditionLocation(finalConditionLocation); } else { - log.info("All conditions are not satisfied, move to the default branch"); moveToDefaultBranch(); } } @@ -147,4 +152,16 @@ public class SwitchLogicTask extends BaseSyncLogicTask { } } + private void addDefaultBranchToEnd() { + SwitchResultVo switchResultVo = new SwitchResultVo(null, taskParameters.getNextNode()); + List dependTaskList = taskParameters.getDependTaskList(); + if (!dependTaskList.contains(switchResultVo)) { + dependTaskList.add(switchResultVo); + } + } + + private SwitchResultVo getDefaultSwitchResultVo() { + return taskParameters.getDependTaskList().get(taskParameters.getDependTaskList().size() - 1); + } + } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtils.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtils.java index 52f274cbcb..1676df7e01 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtils.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtils.java @@ -63,7 +63,6 @@ public class SwitchTaskUtils { if (MapUtils.isNotEmpty(varParams)) { params.putAll(varParams); } - String originContent = content; Pattern pattern = Pattern.compile(rgex); Matcher m = pattern.matcher(content); while (m.find()) { @@ -82,10 +81,6 @@ public class SwitchTaskUtils { content = content.replace("${" + paramName + "}", value); } - // if not replace any params, throw exception to avoid illegal condition - if (originContent.equals(content)) { - throw new IllegalArgumentException("condition is not valid, please check it. condition: " + condition); - } return content; } diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtilsTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtilsTest.java index 7f7ae43bbf..34785ada47 100644 --- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtilsTest.java +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/utils/SwitchTaskUtilsTest.java @@ -50,7 +50,7 @@ public class SwitchTaskUtilsTest { Map globalParams = new HashMap<>(); Map varParams = new HashMap<>(); globalParams.put("test", new Property("test", Direct.IN, DataType.INTEGER, "1")); - Assertions.assertThrowsExactly(IllegalArgumentException.class, () -> { + Assertions.assertDoesNotThrow(() -> { SwitchTaskUtils.generateContentWithTaskParams(content, globalParams, varParams); }); @@ -70,15 +70,5 @@ public class SwitchTaskUtilsTest { SwitchTaskUtils.evaluate(script); }); - String contentWithSpecify1 = "cmd.abc"; - Assertions.assertThrowsExactly(IllegalArgumentException.class, () -> { - SwitchTaskUtils.generateContentWithTaskParams(contentWithSpecify1, globalParams, varParams); - }); - - String contentWithSpecify2 = "cmd()"; - Assertions.assertThrowsExactly(IllegalArgumentException.class, () -> { - SwitchTaskUtils.generateContentWithTaskParams(contentWithSpecify2, globalParams, varParams); - }); - } } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/TaskInstanceLogHeader.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/TaskInstanceLogHeader.java index 02ee2e1c99..3aa76dd114 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/TaskInstanceLogHeader.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/TaskInstanceLogHeader.java @@ -27,11 +27,11 @@ import com.google.common.collect.Lists; public class TaskInstanceLogHeader { private static final List INITIALIZE_TASK_CONTEXT_HEADER = Lists.newArrayList( - "***********************************************************************************************", + "\n***********************************************************************************************", "********************************* Initialize task context ***********************************", "***********************************************************************************************"); private static final List LOAD_TASK_INSTANCE_PLUGIN_HEADER = Lists.newArrayList( - "***********************************************************************************************", + "\n***********************************************************************************************", "********************************* Load task instance plugin *********************************", "***********************************************************************************************"); @@ -40,12 +40,12 @@ public class TaskInstanceLogHeader { } private static final List EXECUTE_TASK_HEADER = Lists.newArrayList( - "***********************************************************************************************", + "\n***********************************************************************************************", "********************************* Execute task instance *************************************", "***********************************************************************************************"); private static final List FINALIZE_TASK_HEADER = Lists.newArrayList( - "***********************************************************************************************", + "\n***********************************************************************************************", "********************************* Finalize task instance ************************************", "***********************************************************************************************");