|
|
@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.common.enums.DependentRelation; |
|
|
|
import org.apache.dolphinscheduler.common.enums.ExecutionStatus; |
|
|
|
import org.apache.dolphinscheduler.common.enums.ExecutionStatus; |
|
|
|
import org.apache.dolphinscheduler.common.model.DateInterval; |
|
|
|
import org.apache.dolphinscheduler.common.model.DateInterval; |
|
|
|
import org.apache.dolphinscheduler.common.model.DependentItem; |
|
|
|
import org.apache.dolphinscheduler.common.model.DependentItem; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.common.model.TaskNode; |
|
|
|
import org.apache.dolphinscheduler.common.utils.DependentUtils; |
|
|
|
import org.apache.dolphinscheduler.common.utils.DependentUtils; |
|
|
|
import org.apache.dolphinscheduler.dao.entity.ProcessInstance; |
|
|
|
import org.apache.dolphinscheduler.dao.entity.ProcessInstance; |
|
|
|
import org.apache.dolphinscheduler.dao.entity.TaskInstance; |
|
|
|
import org.apache.dolphinscheduler.dao.entity.TaskInstance; |
|
|
@ -82,7 +83,7 @@ public class DependentExecute { |
|
|
|
* @param currentTime current time |
|
|
|
* @param currentTime current time |
|
|
|
* @return DependResult |
|
|
|
* @return DependResult |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
public DependResult getDependentResultForItem(DependentItem dependentItem, Date currentTime){ |
|
|
|
private DependResult getDependentResultForItem(DependentItem dependentItem, Date currentTime){ |
|
|
|
List<DateInterval> dateIntervals = DependentUtils.getDateIntervalList(currentTime, dependentItem.getDateValue()); |
|
|
|
List<DateInterval> dateIntervals = DependentUtils.getDateIntervalList(currentTime, dependentItem.getDateValue()); |
|
|
|
return calculateResultForTasks(dependentItem, dateIntervals ); |
|
|
|
return calculateResultForTasks(dependentItem, dateIntervals ); |
|
|
|
} |
|
|
|
} |
|
|
@ -95,6 +96,7 @@ public class DependentExecute { |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private DependResult calculateResultForTasks(DependentItem dependentItem, |
|
|
|
private DependResult calculateResultForTasks(DependentItem dependentItem, |
|
|
|
List<DateInterval> dateIntervals) { |
|
|
|
List<DateInterval> dateIntervals) { |
|
|
|
|
|
|
|
|
|
|
|
DependResult result = DependResult.FAILED; |
|
|
|
DependResult result = DependResult.FAILED; |
|
|
|
for(DateInterval dateInterval : dateIntervals){ |
|
|
|
for(DateInterval dateInterval : dateIntervals){ |
|
|
|
ProcessInstance processInstance = findLastProcessInterval(dependentItem.getDefinitionId(), |
|
|
|
ProcessInstance processInstance = findLastProcessInterval(dependentItem.getDefinitionId(), |
|
|
@ -104,30 +106,69 @@ public class DependentExecute { |
|
|
|
dependentItem.getDefinitionId(), dateInterval.getStartTime(), dateInterval.getEndTime() ); |
|
|
|
dependentItem.getDefinitionId(), dateInterval.getStartTime(), dateInterval.getEndTime() ); |
|
|
|
return DependResult.FAILED; |
|
|
|
return DependResult.FAILED; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// need to check workflow for updates, so get all task and check the task state
|
|
|
|
if(dependentItem.getDepTasks().equals(Constants.DEPENDENT_ALL)){ |
|
|
|
if(dependentItem.getDepTasks().equals(Constants.DEPENDENT_ALL)){ |
|
|
|
result = getDependResultByState(processInstance.getState()); |
|
|
|
List<TaskNode> taskNodes = |
|
|
|
|
|
|
|
processService.getTaskNodeListByDefinitionId(dependentItem.getDefinitionId()); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if(taskNodes != null && taskNodes.size() > 0){ |
|
|
|
|
|
|
|
List<DependResult> results = new ArrayList<>(); |
|
|
|
|
|
|
|
DependResult tmpResult = DependResult.FAILED; |
|
|
|
|
|
|
|
for(TaskNode taskNode:taskNodes){ |
|
|
|
|
|
|
|
tmpResult = getDependTaskResult(taskNode.getName(),processInstance); |
|
|
|
|
|
|
|
if(DependResult.FAILED == tmpResult){ |
|
|
|
|
|
|
|
break; |
|
|
|
|
|
|
|
}else{ |
|
|
|
|
|
|
|
results.add(getDependTaskResult(taskNode.getName(),processInstance)); |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if(DependResult.FAILED == tmpResult){ |
|
|
|
|
|
|
|
result = DependResult.FAILED; |
|
|
|
|
|
|
|
}else if(results.contains(DependResult.WAITING)){ |
|
|
|
|
|
|
|
result = DependResult.WAITING; |
|
|
|
|
|
|
|
}else{ |
|
|
|
|
|
|
|
result = DependResult.SUCCESS; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
}else{ |
|
|
|
|
|
|
|
result = DependResult.FAILED; |
|
|
|
|
|
|
|
} |
|
|
|
}else{ |
|
|
|
}else{ |
|
|
|
|
|
|
|
result = getDependTaskResult(dependentItem.getDepTasks(),processInstance); |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
if(result != DependResult.SUCCESS){ |
|
|
|
|
|
|
|
break; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
return result; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
|
|
|
* get depend task result |
|
|
|
|
|
|
|
* @param taskName |
|
|
|
|
|
|
|
* @param processInstance |
|
|
|
|
|
|
|
* @return |
|
|
|
|
|
|
|
*/ |
|
|
|
|
|
|
|
private DependResult getDependTaskResult(String taskName, ProcessInstance processInstance) { |
|
|
|
|
|
|
|
DependResult result = DependResult.FAILED; |
|
|
|
TaskInstance taskInstance = null; |
|
|
|
TaskInstance taskInstance = null; |
|
|
|
List<TaskInstance> taskInstanceList = processService.findValidTaskListByProcessId(processInstance.getId()); |
|
|
|
List<TaskInstance> taskInstanceList = processService.findValidTaskListByProcessId(processInstance.getId()); |
|
|
|
|
|
|
|
|
|
|
|
for(TaskInstance task : taskInstanceList){ |
|
|
|
for(TaskInstance task : taskInstanceList){ |
|
|
|
if(task.getName().equals(dependentItem.getDepTasks())){ |
|
|
|
if(task.getName().equals(taskName)){ |
|
|
|
taskInstance = task; |
|
|
|
taskInstance = task; |
|
|
|
break; |
|
|
|
break; |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
if(taskInstance == null){ |
|
|
|
if(taskInstance == null){ |
|
|
|
// cannot find task in the process instance
|
|
|
|
// cannot find task in the process instance
|
|
|
|
// maybe because process instance is running or failed.
|
|
|
|
// maybe because process instance is running or failed.
|
|
|
|
result = getDependResultByState(processInstance.getState()); |
|
|
|
result = getDependResultByProcessStateWhenTaskNull(processInstance.getState()); |
|
|
|
}else{ |
|
|
|
}else{ |
|
|
|
result = getDependResultByState(taskInstance.getState()); |
|
|
|
result = getDependResultByState(taskInstance.getState()); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
if(result != DependResult.SUCCESS){ |
|
|
|
|
|
|
|
break; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
return result; |
|
|
|
return result; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -172,7 +213,9 @@ public class DependentExecute { |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private DependResult getDependResultByState(ExecutionStatus state) { |
|
|
|
private DependResult getDependResultByState(ExecutionStatus state) { |
|
|
|
|
|
|
|
|
|
|
|
if(state.typeIsRunning() || state == ExecutionStatus.SUBMITTED_SUCCESS || state == ExecutionStatus.WAITTING_THREAD){ |
|
|
|
if(state.typeIsRunning() |
|
|
|
|
|
|
|
|| state == ExecutionStatus.SUBMITTED_SUCCESS |
|
|
|
|
|
|
|
|| state == ExecutionStatus.WAITTING_THREAD){ |
|
|
|
return DependResult.WAITING; |
|
|
|
return DependResult.WAITING; |
|
|
|
}else if(state.typeIsSuccess()){ |
|
|
|
}else if(state.typeIsSuccess()){ |
|
|
|
return DependResult.SUCCESS; |
|
|
|
return DependResult.SUCCESS; |
|
|
@ -181,6 +224,22 @@ public class DependentExecute { |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
|
|
|
* get dependent result by task instance state when task instance is null |
|
|
|
|
|
|
|
* @param state state |
|
|
|
|
|
|
|
* @return DependResult |
|
|
|
|
|
|
|
*/ |
|
|
|
|
|
|
|
private DependResult getDependResultByProcessStateWhenTaskNull(ExecutionStatus state) { |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if(state.typeIsRunning() |
|
|
|
|
|
|
|
|| state == ExecutionStatus.SUBMITTED_SUCCESS |
|
|
|
|
|
|
|
|| state == ExecutionStatus.WAITTING_THREAD){ |
|
|
|
|
|
|
|
return DependResult.WAITING; |
|
|
|
|
|
|
|
}else{ |
|
|
|
|
|
|
|
return DependResult.FAILED; |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* judge depend item finished |
|
|
|
* judge depend item finished |
|
|
|
* @param currentTime current time |
|
|
|
* @param currentTime current time |
|
|
@ -222,7 +281,7 @@ public class DependentExecute { |
|
|
|
* @param currentTime current time |
|
|
|
* @param currentTime current time |
|
|
|
* @return DependResult |
|
|
|
* @return DependResult |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
public DependResult getDependResultForItem(DependentItem item, Date currentTime){ |
|
|
|
private DependResult getDependResultForItem(DependentItem item, Date currentTime){ |
|
|
|
String key = item.getKey(); |
|
|
|
String key = item.getKey(); |
|
|
|
if(dependResultMap.containsKey(key)){ |
|
|
|
if(dependResultMap.containsKey(key)){ |
|
|
|
return dependResultMap.get(key); |
|
|
|
return dependResultMap.get(key); |
|
|
|