This is an automated email from the ASF dual-hosted git repository.
SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new cd8abbaa62 [Improvement-18249][DAO] Route TaskDefinitionMapper access
through TaskDefinitionDao (#18254)
cd8abbaa62 is described below
commit cd8abbaa626d1b7da229fc6883cbb80b687428d6
Author: Wenjun Ruan <[email protected]>
AuthorDate: Wed May 20 11:42:13 2026 +0800
[Improvement-18249][DAO] Route TaskDefinitionMapper access through
TaskDefinitionDao (#18254)
---
.../audit/operator/impl/TaskAuditOperatorImpl.java | 6 ++--
.../dolphinscheduler/api/python/PythonGateway.java | 8 ++---
.../api/service/impl/EnvironmentServiceImpl.java | 13 ++++-----
.../api/service/impl/ExecutorServiceImpl.java | 6 ++--
.../api/service/impl/LoggerServiceImpl.java | 8 ++---
.../service/impl/TaskDefinitionServiceImpl.java | 34 ++++++++++------------
.../api/service/impl/WorkerGroupServiceImpl.java | 7 ++---
.../impl/WorkflowDefinitionServiceImpl.java | 6 ++--
.../service/impl/WorkflowInstanceServiceImpl.java | 8 ++---
.../service/impl/WorkflowLineageServiceImpl.java | 12 ++++----
.../api/python/PythonGatewayTest.java | 8 ++---
.../api/service/LoggerServiceTest.java | 8 ++---
.../api/service/TaskDefinitionServiceImplTest.java | 24 ++++++++-------
.../api/service/WorkerGroupServiceTest.java | 6 ++--
.../api/service/WorkflowDefinitionServiceTest.java | 4 +--
.../api/service/WorkflowInstanceServiceTest.java | 8 ++---
.../service/WorkflowTaskLineageServiceTest.java | 14 ++++-----
.../api/service/impl/EnvironmentServiceTest.java | 9 +++---
.../impl/WorkflowLineageServiceImplTest.java | 12 ++++----
.../dao/repository/TaskDefinitionDao.java | 8 +++++
.../dao/repository/impl/TaskDefinitionDaoImpl.java | 26 +++++++++++++++++
21 files changed, 132 insertions(+), 103 deletions(-)
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/audit/operator/impl/TaskAuditOperatorImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/audit/operator/impl/TaskAuditOperatorImpl.java
index db6559b865..fd6dd1115c 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/audit/operator/impl/TaskAuditOperatorImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/audit/operator/impl/TaskAuditOperatorImpl.java
@@ -24,7 +24,7 @@ import
org.apache.dolphinscheduler.api.audit.operator.BaseAuditOperator;
import org.apache.dolphinscheduler.common.enums.AuditOperationType;
import org.apache.dolphinscheduler.dao.entity.AuditLog;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import java.util.List;
import java.util.Map;
@@ -36,7 +36,7 @@ import org.springframework.stereotype.Service;
public class TaskAuditOperatorImpl extends BaseAuditOperator {
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Override
public void modifyAuditOperationType(AuditType auditType, Map<String,
Object> paramsMap,
@@ -62,7 +62,7 @@ public class TaskAuditOperatorImpl extends BaseAuditOperator {
return "";
}
- TaskDefinition obj = taskDefinitionMapper.queryByCode(objId);
+ TaskDefinition obj = taskDefinitionDao.queryByCode(objId);
return obj == null ? "" : obj.getName();
}
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
index 906d5605e8..1375ebb127 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
@@ -50,11 +50,11 @@ import
org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.DataSourceDao;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.ProjectUserDao;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.plugin.storage.api.StorageEntity;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
@@ -123,7 +123,7 @@ public class PythonGateway {
private ProjectDao projectDao;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Autowired
private SchedulerService schedulerService;
@@ -192,7 +192,7 @@ public class PythonGateway {
}
TaskDefinition taskDefinition =
- taskDefinitionMapper.queryByName(project.getCode(),
workflowDefinition.getCode(), taskName);
+ taskDefinitionDao.queryByName(project.getCode(),
workflowDefinition.getCode(), taskName);
if (taskDefinition == null) {
result.put("code", CodeGenerateUtils.genCode());
result.put("version", 0L);
@@ -572,7 +572,7 @@ public class PythonGateway {
if (taskName != null) {
TaskDefinition taskDefinition =
- taskDefinitionMapper.queryByName(projectCode,
workflowDefinition.getCode(), taskName);
+ taskDefinitionDao.queryByName(projectCode,
workflowDefinition.getCode(), taskName);
result.put("taskDefinitionCode", taskDefinition.getCode());
}
return result;
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceImpl.java
index 1d96750714..96b04c19b9 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceImpl.java
@@ -37,7 +37,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
import
org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.SetUtils;
@@ -80,7 +80,7 @@ public class EnvironmentServiceImpl extends BaseServiceImpl
implements Environme
private EnvironmentWorkerGroupRelationMapper relationMapper;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
/**
* create environment
@@ -282,8 +282,7 @@ public class EnvironmentServiceImpl extends BaseServiceImpl
implements Environme
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
- Long relatedTaskNumber = taskDefinitionMapper
- .selectCount(new
QueryWrapper<TaskDefinition>().lambda().eq(TaskDefinition::getEnvironmentCode,
code));
+ long relatedTaskNumber =
taskDefinitionDao.countByEnvironmentCode(code);
if (relatedTaskNumber > 0) {
log.warn("Delete environment failed because {} tasks is using it,
environmentCode:{}.",
@@ -407,10 +406,8 @@ public class EnvironmentServiceImpl extends
BaseServiceImpl implements Environme
private void checkUsedEnvironmentWorkerGroupRelation(Set<String>
deleteKeySet,
String
environmentName, Long environmentCode) {
for (String workerGroup : deleteKeySet) {
- List<TaskDefinition> taskDefinitionList = taskDefinitionMapper
- .selectList(new QueryWrapper<TaskDefinition>().lambda()
- .eq(TaskDefinition::getEnvironmentCode,
environmentCode)
- .eq(TaskDefinition::getWorkerGroup, workerGroup));
+ List<TaskDefinition> taskDefinitionList =
+
taskDefinitionDao.queryByEnvironmentCodeAndWorkerGroup(environmentCode,
workerGroup);
if (Objects.nonNull(taskDefinitionList) &&
taskDefinitionList.size() != 0) {
Set<String> collect =
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
index 70d3f084ec..8950462eaf 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -51,8 +51,8 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
@@ -99,7 +99,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl
implements ExecutorServ
private TaskDefinitionLogMapper taskDefinitionLogMapper;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Autowired
private WorkflowTaskRelationDao workflowTaskRelationDao;
@@ -203,7 +203,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl
implements ExecutorServ
}
Set<Long> relationCodes =
workflowTaskRelations.stream().map(WorkflowTaskRelation::getPostTaskCode).collect(Collectors.toSet());
- List<TaskDefinition> taskDefinitions =
taskDefinitionMapper.queryByCodeList(relationCodes);
+ List<TaskDefinition> taskDefinitions =
taskDefinitionDao.queryByCodes(relationCodes);
// find out the workflow definition code
Set<Long> workflowDefinitionCodeSet = new HashSet<>();
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
index 45c0099e08..fac27c9345 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
@@ -32,8 +32,8 @@ import org.apache.dolphinscheduler.dao.entity.ResponseTaskLog;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.commons.lang3.StringUtils;
@@ -63,7 +63,7 @@ public class LoggerServiceImpl extends BaseServiceImpl
implements LoggerService
private ProjectService projectService;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Autowired
private LogClientDelegate logClientDelegate;
@@ -138,7 +138,7 @@ public class LoggerServiceImpl extends BaseServiceImpl
implements LoggerService
throw new ServiceException(Status.TASK_INSTANCE_NOT_FOUND);
}
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(task.getTaskCode());
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(task.getTaskCode());
if (taskDefinition != null && projectCode !=
taskDefinition.getProjectCode()) {
throw new ServiceException(Status.TASK_INSTANCE_NOT_FOUND,
taskInstId);
}
@@ -164,7 +164,7 @@ public class LoggerServiceImpl extends BaseServiceImpl
implements LoggerService
throw new ServiceException("task instance is null or host is
null");
}
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(task.getTaskCode());
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(task.getTaskCode());
if (taskDefinition != null && projectCode !=
taskDefinition.getProjectCode()) {
throw new ServiceException("task instance does not exist in
project");
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
index be2764c873..4828f9acd5 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
@@ -49,7 +49,6 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelationLog;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
@@ -94,9 +93,6 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
@Autowired
private ProjectService projectService;
- @Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
-
@Autowired
private TaskDefinitionDao taskDefinitionDao;
@@ -137,7 +133,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
projectService.checkProjectAndAuthThrowException(loginUser, project,
TASK_DEFINITION);
TaskDefinition taskDefinition =
- taskDefinitionMapper.queryByName(project.getCode(),
workflowDefinitionCode, taskName);
+ taskDefinitionDao.queryByName(project.getCode(),
workflowDefinitionCode, taskName);
if (taskDefinition == null) {
log.error("Task definition does not exist, taskName:{}.",
taskName);
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, taskName);
@@ -189,7 +185,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
@Override
public TaskDefinition getTaskDefinition(User loginUser,
long taskCode) {
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskCode);
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskCode);
if (taskDefinition == null) {
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST, taskCode);
}
@@ -212,7 +208,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
// check if user have write perm for project
projectService.checkHasProjectWritePermissionThrowException(loginUser,
project);
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskCode);
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskCode);
if (taskDefinition == null) {
log.error("Task definition does not exist,
taskDefinitionCode:{}.", taskCode);
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST,
String.valueOf(taskCode));
@@ -255,13 +251,13 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
taskDefinitionToUpdate.setVersion(++version);
taskDefinitionToUpdate.setTaskType(taskDefinitionToUpdate.getTaskType().toUpperCase());
taskDefinitionToUpdate.setUpdateTime(now);
- int update = taskDefinitionMapper.updateById(taskDefinitionToUpdate);
+ boolean updateSuccess =
taskDefinitionDao.updateById(taskDefinitionToUpdate);
taskDefinitionToUpdate.setOperator(loginUser.getId());
taskDefinitionToUpdate.setOperateTime(now);
taskDefinitionToUpdate.setCreateTime(now);
taskDefinitionToUpdate.setId(null);
int insert = taskDefinitionLogMapper.insert(taskDefinitionToUpdate);
- if ((update & insert) != 1) {
+ if (!updateSuccess || insert != 1) {
log.error("Update task definition or definitionLog error,
projectCode:{}, taskDefinitionCode:{}.",
projectCode, taskCode);
throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
@@ -355,7 +351,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
}
Map<Long, TaskDefinition> queryUpStreamTaskCodeMap;
if (CollectionUtils.isNotEmpty(upstreamTaskCodes)) {
- List<TaskDefinition> upstreamTaskDefinitionList =
taskDefinitionMapper.queryByCodeList(upstreamTaskCodes);
+ List<TaskDefinition> upstreamTaskDefinitionList =
taskDefinitionDao.queryByCodes(upstreamTaskCodes);
queryUpStreamTaskCodeMap = upstreamTaskDefinitionList.stream()
.collect(Collectors.toMap(TaskDefinition::getCode,
taskDefinition -> taskDefinition));
// upstreamTaskCodes - queryUpStreamTaskCodeMap.keySet
@@ -447,7 +443,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
}
// get add task code map
allPreTaskCodeSet.add(Long.valueOf(taskCode));
- List<TaskDefinition> taskDefinitionList =
taskDefinitionMapper.queryByCodeList(allPreTaskCodeSet);
+ List<TaskDefinition> taskDefinitionList =
taskDefinitionDao.queryByCodes(allPreTaskCodeSet);
Map<Long, TaskDefinition> taskCodeMap =
taskDefinitionList.stream().collect(Collectors
.toMap(TaskDefinition::getCode, Function.identity(), (a, b) ->
a));
@@ -515,7 +511,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
ReleaseState.ONLINE.getDescp(), taskCode);
throw new ServiceException(Status.WORKFLOW_DEFINE_STATE_ONLINE);
}
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskCode);
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskCode);
if (taskDefinition == null || projectCode !=
taskDefinition.getProjectCode()) {
log.error("Task definition does not exist,
taskDefinitionCode:{}.", taskCode);
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST,
String.valueOf(taskCode));
@@ -525,8 +521,8 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
taskDefinitionUpdate.setUserId(loginUser.getId());
taskDefinitionUpdate.setUpdateTime(new Date());
taskDefinitionUpdate.setId(taskDefinition.getId());
- int switchVersion =
taskDefinitionMapper.updateById(taskDefinitionUpdate);
- if (switchVersion <= 0) {
+ boolean switchSuccess =
taskDefinitionDao.updateById(taskDefinitionUpdate);
+ if (!switchSuccess) {
log.error("Task definition version switch error,
taskDefinitionCode:{}.", taskCode);
throw new
ServiceException(Status.SWITCH_TASK_DEFINITION_VERSION_ERROR);
}
@@ -578,7 +574,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
// check if user have write perm for project
projectService.checkHasProjectWritePermissionThrowException(loginUser,
project);
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskCode);
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskCode);
if (taskDefinition == null || projectCode !=
taskDefinition.getProjectCode()) {
log.error("Task definition does not exist,
taskDefinitionCode:{}.", taskCode);
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST,
String.valueOf(taskCode));
@@ -606,7 +602,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
// check user access for project
projectService.checkProjectAndAuthThrowException(loginUser, project,
TASK_DEFINITION);
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskCode);
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskCode);
if (taskDefinition == null || projectCode !=
taskDefinition.getProjectCode()) {
log.error("Task definition does not exist,
taskDefinitionCode:{}.", taskCode);
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST,
String.valueOf(taskCode));
@@ -653,7 +649,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
if (null == releaseState) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR,
Constants.RELEASE_STATE);
}
- TaskDefinition taskDefinition = taskDefinitionMapper.queryByCode(code);
+ TaskDefinition taskDefinition = taskDefinitionDao.queryByCode(code);
if (taskDefinition == null || projectCode !=
taskDefinition.getProjectCode()) {
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST,
String.valueOf(code));
}
@@ -689,9 +685,9 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
log.warn("Parameter releaseState is invalid.");
throw new
ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR,
Constants.RELEASE_STATE);
}
- int update = taskDefinitionMapper.updateById(taskDefinition);
+ boolean updateSuccess = taskDefinitionDao.updateById(taskDefinition);
int updateLog = taskDefinitionLogMapper.updateById(taskDefinitionLog);
- if ((update == 0 && updateLog == 1) || (update == 1 && updateLog ==
0)) {
+ if (updateSuccess != (updateLog == 1)) {
log.error("Update taskDefinition state or taskDefinitionLog state
error, taskDefinitionCode:{}.", code);
throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
index ff0325d84b..ea7cc151cc 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
@@ -42,8 +42,8 @@ import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
import org.apache.dolphinscheduler.dao.entity.WorkerGroupPageDetail;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
import
org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkerGroupDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
@@ -93,7 +93,7 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl
implements WorkerGro
private ScheduleDao scheduleDao;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Autowired
private WorkflowDefinitionDao workflowDefinitionDao;
@@ -158,8 +158,7 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl
implements WorkerGro
*/
private void checkWorkerGroupDependencies(WorkerGroup workerGroup) {
// check if the worker group has any dependent tasks
- List<TaskDefinition> taskDefinitions = taskDefinitionMapper.selectList(
- new
QueryWrapper<TaskDefinition>().lambda().eq(TaskDefinition::getWorkerGroup,
workerGroup.getName()));
+ List<TaskDefinition> taskDefinitions =
taskDefinitionDao.queryByWorkerGroup(workerGroup.getName());
if (CollectionUtils.isNotEmpty(taskDefinitions)) {
List<String> taskNames =
taskDefinitions.stream().limit(3).map(TaskDefinition::getName)
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
index e883fe99c3..8609057f87 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
@@ -78,12 +78,12 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowTaskLineage;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelationLog;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationLogMapper;
import org.apache.dolphinscheduler.dao.model.PageListingResult;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.repository.UserDao;
@@ -197,7 +197,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
private TaskDefinitionLogService taskDefinitionLogService;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Lazy
@Autowired
@@ -1816,7 +1816,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
}
});
- taskDefinitionMapper.queryByCodeList(taskCodeSet)
+ taskDefinitionDao.queryByCodes(taskCodeSet)
.stream().forEach(taskDefinition -> {
Map<String, Object> localParamsMap = new HashMap<>();
String localParams =
JSONUtils.getNodeString(taskDefinition.getTaskParams(), LOCAL_PARAMS);
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowInstanceServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowInstanceServiceImpl.java
index 7dccef947a..5d024bbb55 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowInstanceServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowInstanceServiceImpl.java
@@ -66,9 +66,9 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelationLog;
import org.apache.dolphinscheduler.dao.mapper.RelationSubWorkflowMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceContextDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
@@ -155,7 +155,7 @@ public class WorkflowInstanceServiceImpl extends
BaseServiceImpl implements Work
UsersService usersService;
@Autowired
- TaskDefinitionMapper taskDefinitionMapper;
+ TaskDefinitionDao taskDefinitionDao;
@Autowired
private RelationSubWorkflowMapper relationSubWorkflowMapper;
@@ -343,7 +343,7 @@ public class WorkflowInstanceServiceImpl extends
BaseServiceImpl implements Work
throw new ServiceException(Status.TASK_INSTANCE_NOT_EXISTS,
taskId);
}
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskInstance.getTaskCode());
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskInstance.getTaskCode());
if (taskDefinition == null) {
throw new ServiceException(Status.TASK_INSTANCE_NOT_EXISTS,
taskId);
}
@@ -399,7 +399,7 @@ public class WorkflowInstanceServiceImpl extends
BaseServiceImpl implements Work
throw new ServiceException(Status.TASK_INSTANCE_NOT_EXISTS,
taskId);
}
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskInstance.getTaskCode());
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskInstance.getTaskCode());
if (taskDefinition != null && projectCode !=
taskDefinition.getProjectCode()) {
log.error("Task definition does not exist, projectCode:{},
taskDefinitionCode:{}.", projectCode,
taskInstance.getTaskCode());
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImpl.java
index cdbeabe92c..44241590d0 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImpl.java
@@ -31,8 +31,8 @@ import
org.apache.dolphinscheduler.dao.entity.WorkFlowRelation;
import org.apache.dolphinscheduler.dao.entity.WorkFlowRelationDetail;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskLineage;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskLineageDao;
@@ -67,7 +67,7 @@ public class WorkflowLineageServiceImpl extends
BaseServiceImpl implements Workf
private ProjectDao projectDao;
@Autowired
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Autowired
private WorkflowTaskLineageDao workflowTaskLineageDao;
@@ -194,7 +194,7 @@ public class WorkflowLineageServiceImpl extends
BaseServiceImpl implements Workf
String taskName = "";
if (workflowTaskLineage.getTaskDefinitionCode() != 0) {
TaskDefinition taskDefinition =
-
taskDefinitionMapper.queryByCode(workflowTaskLineage.getTaskDefinitionCode());
+
taskDefinitionDao.queryByCode(workflowTaskLineage.getTaskDefinitionCode());
// Handle dirty data scenario caused by historical bugs:
// There may be orphaned lineage records referencing deleted
task definitions.
// Skip these records to prevent NPE and ensure the method
continues processing.
@@ -218,7 +218,7 @@ public class WorkflowLineageServiceImpl extends
BaseServiceImpl implements Workf
String taskDepStr = String.join(Constants.COMMA, taskDepStrList);
if (taskCode != 0) {
- TaskDefinition taskDefinition =
taskDefinitionMapper.queryByCode(taskCode);
+ TaskDefinition taskDefinition =
taskDefinitionDao.queryByCode(taskCode);
return Optional
.of(MessageFormat.format(Status.DELETE_TASK_USE_BY_OTHER_FAIL.getMsg(),
taskDefinition.getName(),
taskDepStr));
@@ -249,7 +249,7 @@ public class WorkflowLineageServiceImpl extends
BaseServiceImpl implements Workf
workflowDefinitionDao.queryByCodes(workflowTaskLineageList.stream()
.map(WorkflowTaskLineage::getWorkflowDefinitionCode).distinct()
.collect(Collectors.toList()));
- List<TaskDefinition> taskDefinitionList =
taskDefinitionMapper.queryByCodeList(workflowTaskLineageList.stream()
+ List<TaskDefinition> taskDefinitionList =
taskDefinitionDao.queryByCodes(workflowTaskLineageList.stream()
.map(WorkflowTaskLineage::getTaskDefinitionCode).filter(code
-> code != 0).distinct()
.collect(Collectors.toList()));
@@ -301,7 +301,7 @@ public class WorkflowLineageServiceImpl extends
BaseServiceImpl implements Workf
List<WorkflowDefinition> workflowDefinitionList =
workflowDefinitionDao.queryByCodes(workflowTaskLineageList.stream()
.map(WorkflowTaskLineage::getWorkflowDefinitionCode).distinct().collect(Collectors.toList()));
- List<TaskDefinition> taskDefinitionList =
taskDefinitionMapper.queryByCodeList(workflowTaskLineageList.stream()
+ List<TaskDefinition> taskDefinitionList =
taskDefinitionDao.queryByCodes(workflowTaskLineageList.stream()
.map(WorkflowTaskLineage::getTaskDefinitionCode).filter(code
-> code != 0).distinct()
.collect(Collectors.toList()));
for (WorkflowTaskLineage workflowTaskLineage :
workflowTaskLineageList) {
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
index 32f39c947b..3063986b9f 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
@@ -22,8 +22,8 @@ import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.plugin.storage.api.StorageEntity;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
@@ -55,7 +55,7 @@ public class PythonGatewayTest {
private WorkflowDefinitionDao workflowDefinitionDao;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Mock
private ResourcesService resourcesService;
@@ -70,7 +70,7 @@ public class PythonGatewayTest {
.thenReturn(workflowDefinition);
TaskDefinition taskDefinition = getTestTaskDefinition();
- Mockito.when(taskDefinitionMapper.queryByName(project.getCode(),
workflowDefinition.getCode(),
+ Mockito.when(taskDefinitionDao.queryByName(project.getCode(),
workflowDefinition.getCode(),
taskDefinition.getName())).thenReturn(taskDefinition);
Map<String, Long> result =
pythonGateway.getCodeAndVersion(project.getName(), workflowDefinition.getName(),
@@ -88,7 +88,7 @@ public class PythonGatewayTest {
.thenReturn(workflowDefinition);
TaskDefinition taskDefinition = getTestTaskDefinition();
- Mockito.when(taskDefinitionMapper.queryByName(project.getCode(),
workflowDefinition.getCode(),
+ Mockito.when(taskDefinitionDao.queryByName(project.getCode(),
workflowDefinition.getCode(),
taskDefinition.getName())).thenReturn(taskDefinition);
Map<String, Object> result =
pythonGateway.getDependentInfo(project.getName(), workflowDefinition.getName(),
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
index 7e3b4cf512..e932bd1b6e 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
@@ -40,8 +40,8 @@ import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import java.text.MessageFormat;
@@ -78,7 +78,7 @@ public class LoggerServiceTest {
private ProjectService projectService;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Mock
private LogClientDelegate logClientDelegate;
@@ -211,7 +211,7 @@ public class LoggerServiceTest {
taskInstance.setLogPath("/temp/log");
doNothing().when(projectService).checkProjectAndAuthThrowException(loginUser,
projectCode, VIEW_LOG);
when(taskInstanceDao.queryById(1)).thenReturn(taskInstance);
-
when(taskDefinitionMapper.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
assertDoesNotThrow(() -> loggerService.queryLog(loginUser,
projectCode, 1, 1, 1));
taskDefinition.setProjectCode(10);
@@ -261,7 +261,7 @@ public class LoggerServiceTest {
Status.INTERNAL_SERVER_ERROR_ARGS, () ->
loggerService.getLogBytes(loginUser, projectCode, 1));
when(taskInstanceDao.queryById(1)).thenReturn(taskInstance);
-
when(taskDefinitionMapper.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
when(logClientDelegate.getWholeLogBytes(any())).thenReturn(new
byte[0]);
assertDoesNotThrow(() -> loggerService.getLogBytes(loginUser,
projectCode, 1));
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
index a8affea832..c749e77e70 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
@@ -52,6 +52,7 @@ import
org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationLogMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationLogDao;
@@ -87,6 +88,9 @@ public class TaskDefinitionServiceImplTest {
@Mock
private TaskDefinitionMapper taskDefinitionMapper;
+ @Mock
+ private TaskDefinitionDao taskDefinitionDao;
+
@Mock
private TaskDefinitionLogMapper taskDefinitionLogMapper;
@@ -152,7 +156,7 @@ public class TaskDefinitionServiceImplTest {
Mockito.doNothing().when(projectService)
.checkProjectAndAuthThrowException(user, project,
TASK_DEFINITION);
- when(taskDefinitionMapper.queryByName(project.getCode(),
PROCESS_DEFINITION_CODE, taskName))
+ when(taskDefinitionDao.queryByName(project.getCode(),
PROCESS_DEFINITION_CODE, taskName))
.thenReturn(new TaskDefinition());
TaskDefinition taskDefinition = taskDefinitionService
@@ -172,9 +176,9 @@ public class TaskDefinitionServiceImplTest {
.thenReturn(new TaskDefinitionLog());
TaskDefinition taskDefinition = new TaskDefinition();
taskDefinition.setProjectCode(PROJECT_CODE);
- when(taskDefinitionMapper.queryByCode(TASK_CODE))
+ when(taskDefinitionDao.queryByCode(TASK_CODE))
.thenReturn(taskDefinition);
-
when(taskDefinitionMapper.updateById(any(TaskDefinitionLog.class))).thenReturn(1);
+
when(taskDefinitionDao.updateById(any(TaskDefinitionLog.class))).thenReturn(true);
Assertions.assertDoesNotThrow(
() -> taskDefinitionService.switchVersion(user, PROJECT_CODE,
TASK_CODE, VERSION));
@@ -191,7 +195,7 @@ public class TaskDefinitionServiceImplTest {
otherProjectTask.setProjectCode(PROJECT_CODE + 1);
otherProjectTask.setCode(TASK_CODE);
otherProjectTask.setVersion(VERSION + 1);
-
when(taskDefinitionMapper.queryByCode(TASK_CODE)).thenReturn(otherProjectTask);
+
when(taskDefinitionDao.queryByCode(TASK_CODE)).thenReturn(otherProjectTask);
assertThrowsServiceException(Status.TASK_DEFINE_NOT_EXIST,
() -> taskDefinitionService.deleteByCodeAndVersion(user,
PROJECT_CODE, TASK_CODE, VERSION));
Mockito.verify(taskDefinitionLogMapper,
Mockito.never()).deleteByCodeAndVersion(TASK_CODE, VERSION);
@@ -201,7 +205,7 @@ public class TaskDefinitionServiceImplTest {
taskDefinition.setProjectCode(PROJECT_CODE);
taskDefinition.setCode(TASK_CODE);
taskDefinition.setVersion(VERSION + 1);
-
when(taskDefinitionMapper.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
when(taskDefinitionLogMapper.deleteByCodeAndVersion(TASK_CODE,
VERSION)).thenReturn(1);
Assertions.assertDoesNotThrow(
() -> taskDefinitionService.deleteByCodeAndVersion(user,
PROJECT_CODE, TASK_CODE, VERSION));
@@ -266,7 +270,7 @@ public class TaskDefinitionServiceImplTest {
"{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"echo
1\",\"conditionResult\":{\"successNode\":[\"\"],\"failedNode\":[\"\"]},\"dependence\":{}}";
taskDefinition.setTaskParams(params);
taskDefinition.setTaskType("SHELL");
-
when(taskDefinitionMapper.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
TaskDefinitionLog taskDefinitionLog = new
TaskDefinitionLog(taskDefinition);
when(taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(TASK_CODE,
taskDefinition.getVersion()))
.thenReturn(taskDefinitionLog);
@@ -328,7 +332,7 @@ public class TaskDefinitionServiceImplTest {
assertEquals(Status.TASK_DEFINE_NOT_EXIST.getCode(),
((ServiceException) exception).getCode());
// error task definition not exists
-
when(taskDefinitionMapper.queryByCode(TASK_CODE)).thenReturn(getTaskDefinition());
+
when(taskDefinitionDao.queryByCode(TASK_CODE)).thenReturn(getTaskDefinition());
when(projectDao.queryByCode(PROJECT_CODE)).thenReturn(getProject());
doThrow(new
ServiceException(Status.USER_NO_OPERATION_PROJECT_PERM)).when(projectService)
.checkProjectAndAuthThrowException(user, getProject(),
TASK_DEFINITION);
@@ -360,12 +364,12 @@ public class TaskDefinitionServiceImplTest {
when(projectDao.queryByCode(PROJECT_CODE)).thenReturn(getProject());
Mockito.doNothing().when(projectService).checkHasProjectWritePermissionThrowException(eq(user),
eq(getProject()));
-
when(taskDefinitionMapper.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(TASK_CODE)).thenReturn(taskDefinition);
when(taskDefinitionLogMapper.queryMaxVersionForDefinition(TASK_CODE)).thenReturn(1);
- when(taskDefinitionMapper.updateById(Mockito.any())).thenReturn(1);
+ when(taskDefinitionDao.updateById(Mockito.any())).thenReturn(true);
when(taskDefinitionLogMapper.insert(Mockito.any())).thenReturn(1);
- when(taskDefinitionMapper.queryByCodeList(Mockito.anySet()))
+ when(taskDefinitionDao.queryByCodes(Mockito.anySet()))
.thenReturn(Arrays.asList(taskDefinition,
taskDefinitionSecond));
when(workflowTaskRelationDao.queryUpstreamByCode(PROJECT_CODE,
TASK_CODE))
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
index e0bc7f22f0..e2ba1b7e6a 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
@@ -36,8 +36,8 @@ import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
import
org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkerGroupDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
import org.apache.dolphinscheduler.registry.api.RegistryClient;
@@ -90,7 +90,7 @@ public class WorkerGroupServiceTest {
private EnvironmentWorkerGroupRelationMapper
environmentWorkerGroupRelationMapper;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Mock
private ScheduleDao scheduleDao;
@@ -253,7 +253,7 @@ public class WorkerGroupServiceTest {
when(workflowInstanceDao.queryByWorkerGroupNameAndStatus(workerGroup.getName(),
WorkflowExecutionStatus.NOT_TERMINAL_STATES)).thenReturn(null);
- when(taskDefinitionMapper.selectList(Mockito.any())).thenReturn(null);
+
when(taskDefinitionDao.queryByWorkerGroup(Mockito.any())).thenReturn(null);
when(scheduleDao.queryScheduleByWorkerGroup(Mockito.any())).thenReturn(null);
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
index b7654c5427..f3d129d8d3 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
@@ -61,12 +61,12 @@ import
org.apache.dolphinscheduler.dao.entity.UserWithWorkflowDefinitionCode;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationLogMapper;
import org.apache.dolphinscheduler.dao.model.PageListingResult;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.repository.UserDao;
@@ -189,7 +189,7 @@ public class WorkflowDefinitionServiceTest extends
BaseServiceTestTool {
private TaskDefinitionLogService taskDefinitionLogService;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Mock
private WorkflowLineageService workflowLineageService;
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowInstanceServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowInstanceServiceTest.java
index b0f6edf511..88e9051ffb 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowInstanceServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowInstanceServiceTest.java
@@ -57,9 +57,9 @@ import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceContextDao;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.repository.TenantDao;
@@ -139,7 +139,7 @@ public class WorkflowInstanceServiceTest {
TenantDao tenantDao;
@Mock
- TaskDefinitionMapper taskDefinitionMapper;
+ TaskDefinitionDao taskDefinitionDao;
@Mock
private TaskInstanceContextDao taskInstanceContextDao;
@@ -496,7 +496,7 @@ public class WorkflowInstanceServiceTest {
when(taskInstanceDao.queryById(1)).thenReturn(taskInstance);
TaskDefinition taskDefinition = new TaskDefinition();
taskDefinition.setProjectCode(projectCode);
-
when(taskDefinitionMapper.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
assertThrowsServiceException(Status.TASK_INSTANCE_NOT_SUB_WORKFLOW_INSTANCE,
() ->
workflowInstanceService.querySubWorkflowInstanceByTaskId(loginUser,
projectCode, 1));
@@ -505,7 +505,7 @@ public class WorkflowInstanceServiceTest {
() ->
workflowInstanceService.querySubWorkflowInstanceByTaskId(loginUser,
projectCode, 1));
taskDefinition.setProjectCode(projectCode);
-
when(taskDefinitionMapper.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
// sub process not exist
TaskInstance subTask = getTaskInstance();
subTask.setTaskType("SUB_WORKFLOW");
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowTaskLineageServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowTaskLineageServiceTest.java
index 3b08338cbf..f391360f4a 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowTaskLineageServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowTaskLineageServiceTest.java
@@ -32,8 +32,8 @@ import
org.apache.dolphinscheduler.dao.entity.WorkFlowRelation;
import org.apache.dolphinscheduler.dao.entity.WorkFlowRelationDetail;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskLineage;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskLineageDao;
@@ -64,7 +64,7 @@ public class WorkflowTaskLineageServiceTest {
private ProjectDao projectDao;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Mock
private WorkflowDefinitionDao workflowDefinitionDao;
@@ -162,7 +162,7 @@ public class WorkflowTaskLineageServiceTest {
when(workflowTaskLineageDao.queryWorkFlowLineageByDept(projectCode,
workflowDefinitionCode, taskCode))
.thenReturn(dependentWorkflowList);
when(workflowDefinitionDao.queryByCode(50L)).thenReturn(Optional.of(workflowDefinition));
- when(taskDefinitionMapper.queryByCode(999L)).thenReturn(null); // Task
definition not found (dirty data)
+ when(taskDefinitionDao.queryByCode(999L)).thenReturn(null); // Task
definition not found (dirty data)
// Should return Optional.empty() because all records are orphaned
Optional<String> result =
@@ -209,13 +209,13 @@ public class WorkflowTaskLineageServiceTest {
.thenReturn(dependentWorkflowList);
when(workflowDefinitionDao.queryByCode(50L)).thenReturn(Optional.of(workflowDefinition1));
when(workflowDefinitionDao.queryByCode(60L)).thenReturn(Optional.of(workflowDefinition2));
-
when(taskDefinitionMapper.queryByCode(300L)).thenReturn(validTaskDefinition);
- when(taskDefinitionMapper.queryByCode(999L)).thenReturn(null); //
Orphaned record
+
when(taskDefinitionDao.queryByCode(300L)).thenReturn(validTaskDefinition);
+ when(taskDefinitionDao.queryByCode(999L)).thenReturn(null); //
Orphaned record
TaskDefinition taskDefinition = new TaskDefinition();
taskDefinition.setCode(taskCode);
taskDefinition.setName("TestTask");
-
when(taskDefinitionMapper.queryByCode(taskCode)).thenReturn(taskDefinition);
+
when(taskDefinitionDao.queryByCode(taskCode)).thenReturn(taskDefinition);
// Should return a message with only the valid record, skipping the
orphaned one
Optional<String> result =
@@ -249,7 +249,7 @@ public class WorkflowTaskLineageServiceTest {
when(workflowTaskLineageDao.queryWorkFlowLineageByDept(projectCode,
workflowDefinitionCode, 0L))
.thenReturn(dependentWorkflowList);
when(workflowDefinitionDao.queryByCode(50L)).thenReturn(Optional.of(workflowDefinition));
- when(taskDefinitionMapper.queryByCode(999L)).thenReturn(null); // Task
definition not found
+ when(taskDefinitionDao.queryByCode(999L)).thenReturn(null); // Task
definition not found
// Should return Optional.empty() because all records are orphaned
Optional<String> result =
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceTest.java
index 4304ecf874..e7cd2a7b43 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/EnvironmentServiceTest.java
@@ -41,7 +41,7 @@ import
org.apache.dolphinscheduler.dao.entity.EnvironmentWorkerGroupRelation;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
import
org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.commons.collections4.CollectionUtils;
@@ -66,7 +66,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -87,7 +86,7 @@ public class EnvironmentServiceTest {
private EnvironmentWorkerGroupRelationMapper relationMapper;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@Mock
private ResourcePermissionCheckService resourcePermissionCheckService;
@@ -282,11 +281,11 @@ public class EnvironmentServiceTest {
() -> environmentService.deleteEnvironmentByCode(generalUser,
1L));
User adminUser = getAdminUser();
-
when(taskDefinitionMapper.selectCount(any(LambdaQueryWrapper.class))).thenReturn(1L);
+ when(taskDefinitionDao.countByEnvironmentCode(1L)).thenReturn(1L);
assertThrowsServiceException(Status.DELETE_ENVIRONMENT_RELATED_TASK_EXISTS,
() -> environmentService.deleteEnvironmentByCode(adminUser,
1L));
-
when(taskDefinitionMapper.selectCount(any(LambdaQueryWrapper.class))).thenReturn(0L);
+ when(taskDefinitionDao.countByEnvironmentCode(1L)).thenReturn(0L);
when(environmentMapper.deleteByCode(1L)).thenReturn(1);
assertDoesNotThrow(() ->
environmentService.deleteEnvironmentByCode(adminUser, 1L));
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImplTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImplTest.java
index 95fc8e036c..e8551c5fd7 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImplTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImplTest.java
@@ -28,7 +28,7 @@ import
org.apache.dolphinscheduler.dao.entity.DependentWorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskLineage;
-import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
+import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskLineageDao;
@@ -57,13 +57,13 @@ class WorkflowLineageServiceImplTest {
private WorkflowDefinitionDao workflowDefinitionDao;
@Mock
- private TaskDefinitionMapper taskDefinitionMapper;
+ private TaskDefinitionDao taskDefinitionDao;
@BeforeEach
void setUp() {
ReflectionTestUtils.setField(workflowLineageService,
"workflowTaskLineageDao", workflowTaskLineageDao);
ReflectionTestUtils.setField(workflowLineageService,
"workflowDefinitionDao", workflowDefinitionDao);
- ReflectionTestUtils.setField(workflowLineageService,
"taskDefinitionMapper", taskDefinitionMapper);
+ ReflectionTestUtils.setField(workflowLineageService,
"taskDefinitionDao", taskDefinitionDao);
}
@Test
@@ -77,7 +77,7 @@ class WorkflowLineageServiceImplTest {
workflowLineageService.queryDownstreamDependentWorkflowDefinitions(workflowCode);
assertThat(result).isEmpty();
- verifyNoInteractions(workflowDefinitionDao, taskDefinitionMapper);
+ verifyNoInteractions(workflowDefinitionDao, taskDefinitionDao);
}
@Test
@@ -116,7 +116,7 @@ class WorkflowLineageServiceImplTest {
taskDefinition.setTaskParams("task-params");
taskDefinition.setWorkerGroup("test-group");
-
when(taskDefinitionMapper.queryByCodeList(Collections.singletonList(300L)))
+ when(taskDefinitionDao.queryByCodes(Collections.singletonList(300L)))
.thenReturn(Collections.singletonList(taskDefinition));
List<DependentWorkflowDefinition> result =
@@ -142,7 +142,7 @@ class WorkflowLineageServiceImplTest {
assertThat(workflowDependent.getWorkerGroup()).isNull();
assertThat(workflowDependent.getWorkflowDefinitionVersion()).isEqualTo(4);
-
verify(taskDefinitionMapper).queryByCodeList(Collections.singletonList(300L));
+
verify(taskDefinitionDao).queryByCodes(Collections.singletonList(300L));
}
@Test
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskDefinitionDao.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskDefinitionDao.java
index 8eb5856c43..e55d249885 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskDefinitionDao.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/TaskDefinitionDao.java
@@ -57,5 +57,13 @@ public interface TaskDefinitionDao extends
IDao<TaskDefinition> {
*/
TaskDefinition queryByCode(long taskCode);
+ TaskDefinition queryByName(long projectCode, long workflowDefinitionCode,
String taskName);
+
+ List<TaskDefinition> queryByWorkerGroup(String workerGroup);
+
+ long countByEnvironmentCode(long environmentCode);
+
+ List<TaskDefinition> queryByEnvironmentCodeAndWorkerGroup(long
environmentCode, String workerGroup);
+
List<String> queryAllTaskDefinitionWorkerGroups(long projectCode);
}
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskDefinitionDaoImpl.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskDefinitionDaoImpl.java
index ca3c9b2e6f..5b54a0e508 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskDefinitionDaoImpl.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/TaskDefinitionDaoImpl.java
@@ -43,6 +43,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.collect.Lists;
@Repository
@@ -117,6 +118,31 @@ public class TaskDefinitionDaoImpl extends
BaseDao<TaskDefinition, TaskDefinitio
return mybatisMapper.queryByCode(taskCode);
}
+ @Override
+ public TaskDefinition queryByName(long projectCode, long
workflowDefinitionCode, String taskName) {
+ return mybatisMapper.queryByName(projectCode, workflowDefinitionCode,
taskName);
+ }
+
+ @Override
+ public List<TaskDefinition> queryByWorkerGroup(String workerGroup) {
+ return mybatisMapper.selectList(
+ new
QueryWrapper<TaskDefinition>().lambda().eq(TaskDefinition::getWorkerGroup,
workerGroup));
+ }
+
+ @Override
+ public long countByEnvironmentCode(long environmentCode) {
+ return mybatisMapper.selectCount(
+ new
QueryWrapper<TaskDefinition>().lambda().eq(TaskDefinition::getEnvironmentCode,
environmentCode));
+ }
+
+ @Override
+ public List<TaskDefinition> queryByEnvironmentCodeAndWorkerGroup(long
environmentCode, String workerGroup) {
+ return mybatisMapper.selectList(
+ new QueryWrapper<TaskDefinition>().lambda()
+ .eq(TaskDefinition::getEnvironmentCode,
environmentCode)
+ .eq(TaskDefinition::getWorkerGroup, workerGroup));
+ }
+
@Override
public List<String> queryAllTaskDefinitionWorkerGroups(long projectCode) {
return mybatisMapper.queryAllTaskDefinitionWorkerGroups(projectCode);
