Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Cherry-pick-3.0.1] [Bug](dependent) Dependent downstream trigger error when schedule cycle not day. #11987

Merged
merged 1 commit into from
Sep 17, 2022
Merged

[Cherry-pick-3.0.1] [Bug](dependent) Dependent downstream trigger error when schedule cycle not day. #11987

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions ...r-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -804,6 +804,7 @@ protected int createComplementDependentCommand(List<Schedule> schedules, Command
dependentCommand.setTaskDependType(TaskDependType.TASK_POST);
for (DependentProcessDefinition dependentProcessDefinition : dependentProcessDefinitionList) {
dependentCommand.setProcessDefinitionCode(dependentProcessDefinition.getProcessDefinitionCode());
dependentCommand.setProcessDefinitionVersion(dependentProcessDefinition.getProcessDefinitionVersion());
dependentCommand.setWorkerGroup(dependentProcessDefinition.getWorkerGroup());
Map<String, String> cmdParam = JSONUtils.toMap(dependentCommand.getCommandParam());
cmdParam.put(CMD_PARAM_START_NODES, String.valueOf(dependentProcessDefinition.getTaskDefinitionCode()));
Expand All @@ -823,17 +824,20 @@ private List<DependentProcessDefinition> getComplementDependentDefinitionList(lo
List<DependentProcessDefinition> dependentProcessDefinitionList =
processService.queryDependentProcessDefinitionByProcessDefinitionCode(processDefinitionCode);

return checkDependentProcessDefinitionValid(dependentProcessDefinitionList,processDefinitionCycle,workerGroup);
return checkDependentProcessDefinitionValid(dependentProcessDefinitionList, processDefinitionCycle,
workerGroup, processDefinitionCode);
}

/**
* Check whether the dependency cycle of the dependent node is consistent with the schedule cycle of
* the dependent process definition and if there is no worker group in the schedule, use the complement selection's
* worker group
*/
private List<DependentProcessDefinition> checkDependentProcessDefinitionValid(List<DependentProcessDefinition> dependentProcessDefinitionList,
CycleEnum processDefinitionCycle,
String workerGroup) {
private List<DependentProcessDefinition> checkDependentProcessDefinitionValid(
List<DependentProcessDefinition> dependentProcessDefinitionList,
CycleEnum processDefinitionCycle,
String workerGroup,
long upstreamProcessDefinitionCode) {
List<DependentProcessDefinition> validDependentProcessDefinitionList = new ArrayList<>();

List<Long> processDefinitionCodeList = dependentProcessDefinitionList.stream()
Expand All @@ -843,8 +847,9 @@ private List<DependentProcessDefinition> checkDependentProcessDefinitionValid(Li
Map<Long, String> processDefinitionWorkerGroupMap = processService.queryWorkerGroupByProcessDefinitionCodes(processDefinitionCodeList);

for (DependentProcessDefinition dependentProcessDefinition : dependentProcessDefinitionList) {
if (dependentProcessDefinition.getDependentCycle() == processDefinitionCycle) {
if (processDefinitionWorkerGroupMap.get(dependentProcessDefinition.getProcessDefinitionCode()) == null) {
if (dependentProcessDefinition.getDependentCycle(upstreamProcessDefinitionCode) == processDefinitionCycle) {
if (processDefinitionWorkerGroupMap
.get(dependentProcessDefinition.getProcessDefinitionCode()) == null) {
dependentProcessDefinition.setWorkerGroup(workerGroup);
}

Expand Down
17 changes: 15 additions & 2 deletions ...-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/DependentProcessDefinition.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,11 @@ public class DependentProcessDefinition {
*/
private String processDefinitionName;

/**
* process definition version
**/
private int processDefinitionVersion;

/**
* task definition name
*/
Expand All @@ -60,14 +65,14 @@ public class DependentProcessDefinition {
* get dependent cycle
* @return CycleEnum
*/
public CycleEnum getDependentCycle() {
public CycleEnum getDependentCycle(long upstreamProcessDefinitionCode) {
DependentParameters dependentParameters = this.getDependentParameters();
List<DependentTaskModel> dependentTaskModelList = dependentParameters.getDependTaskList();

for (DependentTaskModel dependentTaskModel : dependentTaskModelList) {
List<DependentItem> dependentItemList = dependentTaskModel.getDependItemList();
for (DependentItem dependentItem : dependentItemList) {
if (this.getProcessDefinitionCode() == dependentItem.getDefinitionCode()) {
if (upstreamProcessDefinitionCode == dependentItem.getDefinitionCode()) {
return cycle2CycleEnum(dependentItem.getCycle());
}
}
Expand Down Expand Up @@ -122,6 +127,14 @@ public void setProcessDefinitionCode(long code) {
this.processDefinitionCode = code;
}

public int getProcessDefinitionVersion() {
return processDefinitionVersion;
}

public void setProcessDefinitionVersion(int processDefinitionVersion) {
this.processDefinitionVersion = processDefinitionVersion;
}

public long getTaskDefinitionCode() {
return this.taskDefinitionCode;
}
Expand Down
1 change: 1 addition & 0 deletions ...r-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkFlowLineageMapper.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,7 @@
SELECT
c.code AS process_definition_code
,c.name AS process_definition_name
,c.version as process_definition_version
,a.code AS task_definition_code
,a.task_params
FROM
Expand Down