You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ke...@apache.org on 2023/01/03 08:29:16 UTC
[dolphinscheduler] branch 3.1.3-prepare updated (aa18878122 -> 543626aa7a)
This is an automated email from the ASF dual-hosted git repository.
kerwin pushed a change to branch 3.1.3-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
from aa18878122 fix sagemaker sync status (#13264)
new 9564bdeefb cherry-pick Fix task group cannot release when kill task #13314
new 543626aa7a [fix][python] schedule unexpect offline each time update from api #13301
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/controller/SchedulerController.java | 10 +-
.../dolphinscheduler/api/python/PythonGateway.java | 2 +
.../api/service/SchedulerService.java | 9 +-
.../api/service/impl/SchedulerServiceImpl.java | 63 ++++++------
.../api/controller/SchedulerControllerTest.java | 10 +-
.../api/service/SchedulerServiceTest.java | 107 ++++++++++++---------
.../master/runner/WorkflowExecuteRunnable.java | 45 ++++++---
.../master/runner/task/CommonTaskProcessor.java | 10 +-
.../service/process/ProcessServiceImpl.java | 8 +-
.../plugin/task/api/utils/LogUtils.java | 45 ++++++++-
10 files changed, 193 insertions(+), 116 deletions(-)
[dolphinscheduler] 02/02: [fix][python] schedule unexpect offline each time update from api #13301
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kerwin pushed a commit to branch 3.1.3-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit 543626aa7a0f40a76cd30085ebfb9c5975bfee51
Author: Jay Chung <zh...@gmail.com>
AuthorDate: Fri Dec 30 09:59:16 2022 +0800
[fix][python] schedule unexpect offline each time update from api #13301
---
.../api/controller/SchedulerController.java | 10 +-
.../dolphinscheduler/api/python/PythonGateway.java | 2 +
.../api/service/SchedulerService.java | 9 +-
.../api/service/impl/SchedulerServiceImpl.java | 63 ++++++------
.../api/controller/SchedulerControllerTest.java | 10 +-
.../api/service/SchedulerServiceTest.java | 107 ++++++++++++---------
6 files changed, 107 insertions(+), 94 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java
index 95c0ad1a6e..33ea065127 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/SchedulerController.java
@@ -185,8 +185,8 @@ public class SchedulerController extends BaseController {
public Result publishScheduleOnline(@ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser,
@ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@PathVariable("id") Integer id) {
- Map<String, Object> result = schedulerService.setScheduleState(loginUser, projectCode, id, ReleaseState.ONLINE);
- return returnDataList(result);
+ schedulerService.setScheduleState(loginUser, projectCode, id, ReleaseState.ONLINE);
+ return Result.success();
}
/**
@@ -207,10 +207,8 @@ public class SchedulerController extends BaseController {
public Result offlineSchedule(@ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser,
@ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@PathVariable("id") Integer id) {
-
- Map<String, Object> result =
- schedulerService.setScheduleState(loginUser, projectCode, id, ReleaseState.OFFLINE);
- return returnDataList(result);
+ schedulerService.setScheduleState(loginUser, projectCode, id, ReleaseState.OFFLINE);
+ return Result.success();
}
/**
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 4ad807bb1b..b858e14f8e 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
@@ -338,6 +338,8 @@ public class PythonGateway {
schedulerService.updateSchedule(user, projectCode, scheduleId, schedule, WarningType.valueOf(warningType),
warningGroupId, DEFAULT_FAILURE_STRATEGY, DEFAULT_PRIORITY, workerGroup, DEFAULT_ENVIRONMENT_CODE);
}
+ // Always set workflow online to set schedule online
+ processDefinitionService.releaseProcessDefinition(user, projectCode, workflowCode, ReleaseState.ONLINE);
schedulerService.setScheduleState(user, projectCode, scheduleId, ReleaseState.ONLINE);
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java
index 70d77690f0..d3ea5a6b42 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java
@@ -92,12 +92,11 @@ public interface SchedulerService {
* @param projectCode project code
* @param id scheduler id
* @param scheduleStatus schedule status
- * @return publish result code
*/
- Map<String, Object> setScheduleState(User loginUser,
- long projectCode,
- Integer id,
- ReleaseState scheduleStatus);
+ void setScheduleState(User loginUser,
+ long projectCode,
+ Integer id,
+ ReleaseState scheduleStatus);
/**
* query schedule
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
index 51002a7aa4..c3eada0bea 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/SchedulerServiceImpl.java
@@ -279,52 +279,48 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
*/
@Override
@Transactional
- public Map<String, Object> setScheduleState(User loginUser,
- long projectCode,
- Integer id,
- ReleaseState scheduleStatus) {
- Map<String, Object> result = new HashMap<>();
-
+ public void setScheduleState(User loginUser,
+ long projectCode,
+ Integer id,
+ ReleaseState scheduleStatus) {
Project project = projectMapper.queryByCode(projectCode);
// check project auth
- boolean hasProjectAndPerm = projectService.hasProjectAndPerm(loginUser, project, result, null);
- if (!hasProjectAndPerm) {
- return result;
- }
+ projectService.checkProjectAndAuthThrowException(loginUser, project, null);
// check schedule exists
Schedule scheduleObj = scheduleMapper.selectById(id);
if (scheduleObj == null) {
- putMsg(result, Status.SCHEDULE_CRON_NOT_EXISTS, id);
- return result;
+ logger.error("Schedule does not exist, scheduleId:{}.", id);
+ throw new ServiceException(Status.SCHEDULE_CRON_NOT_EXISTS, id);
}
// check schedule release state
if (scheduleObj.getReleaseState() == scheduleStatus) {
- logger.info("schedule release is already {},needn't to change schedule id: {} from {} to {}",
- scheduleObj.getReleaseState(), scheduleObj.getId(), scheduleObj.getReleaseState(), scheduleStatus);
- putMsg(result, Status.SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE, scheduleStatus);
- return result;
+ logger.warn("Schedule state does not need to change due to schedule state is already {}, scheduleId:{}.",
+ scheduleObj.getReleaseState().getDescp(), scheduleObj.getId());
+ throw new ServiceException(Status.SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE, scheduleStatus);
}
ProcessDefinition processDefinition =
processDefinitionMapper.queryByCode(scheduleObj.getProcessDefinitionCode());
if (processDefinition == null || projectCode != processDefinition.getProjectCode()) {
- putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, String.valueOf(scheduleObj.getProcessDefinitionCode()));
- return result;
+ logger.error("Process definition does not exist, processDefinitionCode:{}.",
+ scheduleObj.getProcessDefinitionCode());
+ throw new ServiceException(Status.PROCESS_DEFINE_NOT_EXIST,
+ String.valueOf(scheduleObj.getProcessDefinitionCode()));
}
List<ProcessTaskRelation> processTaskRelations =
processTaskRelationMapper.queryByProcessCode(projectCode, scheduleObj.getProcessDefinitionCode());
if (processTaskRelations.isEmpty()) {
- putMsg(result, Status.PROCESS_DAG_IS_EMPTY);
- return result;
+ logger.error("Process task relations do not exist, projectCode:{}, processDefinitionCode:{}.", projectCode,
+ processDefinition.getCode());
+ throw new ServiceException(Status.PROCESS_DAG_IS_EMPTY);
}
if (scheduleStatus == ReleaseState.ONLINE) {
// check process definition release state
if (processDefinition.getReleaseState() != ReleaseState.ONLINE) {
- logger.info("not release process definition id: {} , name : {}", processDefinition.getId(),
- processDefinition.getName());
- putMsg(result, Status.PROCESS_DEFINE_NOT_RELEASE, processDefinition.getName());
- return result;
+ logger.warn("Only process definition state is {} can change schedule state, processDefinitionCode:{}.",
+ ReleaseState.ONLINE.getDescp(), processDefinition.getCode());
+ throw new ServiceException(Status.PROCESS_DEFINE_NOT_RELEASE, processDefinition.getName());
}
// check sub process definition release state
List<Long> subProcessDefineCodes = new ArrayList<>();
@@ -338,11 +334,11 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
* if there is no online process, exit directly
*/
if (subProcessDefinition.getReleaseState() != ReleaseState.ONLINE) {
- logger.info("not release process definition id: {} , name : {}",
- subProcessDefinition.getId(), subProcessDefinition.getName());
- putMsg(result, Status.PROCESS_DEFINE_NOT_RELEASE,
+ logger.warn(
+ "Only sub process definition state is {} can change schedule state, subProcessDefinitionCode:{}.",
+ ReleaseState.ONLINE.getDescp(), subProcessDefinition.getCode());
+ throw new ServiceException(Status.PROCESS_DEFINE_NOT_RELEASE,
String.valueOf(subProcessDefinition.getId()));
- return result;
}
}
}
@@ -353,8 +349,8 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
List<Server> masterServers = monitorService.getServerListFromRegistry(true);
if (masterServers.isEmpty()) {
- putMsg(result, Status.MASTER_NOT_EXISTS);
- return result;
+ logger.error("Master does not exist.");
+ throw new ServiceException(Status.MASTER_NOT_EXISTS);
}
// set status
@@ -375,18 +371,13 @@ public class SchedulerServiceImpl extends BaseServiceImpl implements SchedulerSe
deleteSchedule(project.getId(), id);
break;
default:
- putMsg(result, Status.SCHEDULE_STATUS_UNKNOWN, scheduleStatus.toString());
- return result;
+ throw new ServiceException(Status.SCHEDULE_STATUS_UNKNOWN, scheduleStatus.toString());
}
} catch (Exception e) {
Status status = scheduleStatus == ReleaseState.ONLINE ? Status.PUBLISH_SCHEDULE_ONLINE_ERROR
: Status.OFFLINE_SCHEDULE_ERROR;
- result.put(Constants.STATUS, status);
throw new ServiceException(status, e);
}
-
- putMsg(result, Status.SUCCESS);
- return result;
}
/**
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/SchedulerControllerTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/SchedulerControllerTest.java
index 7b90330b3d..1dcc56f2a6 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/SchedulerControllerTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/SchedulerControllerTest.java
@@ -119,8 +119,9 @@ public class SchedulerControllerTest extends AbstractControllerTest {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("id","37");
- Mockito.when(schedulerService.setScheduleState(isA(User.class), isA(Long.class), isA(Integer.class),
- isA(ReleaseState.class))).thenReturn(success());
+ Mockito.doNothing().when(schedulerService).setScheduleState(isA(User.class), isA(Long.class),
+ isA(Integer.class),
+ isA(ReleaseState.class));
MvcResult mvcResult = mockMvc.perform(post("/projects/{projectCode}/schedules/{id}/online",123, 37)
.header(SESSION_ID, sessionId)
@@ -139,8 +140,9 @@ public class SchedulerControllerTest extends AbstractControllerTest {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("id","28");
- Mockito.when(schedulerService.setScheduleState(isA(User.class), isA(Long.class), isA(Integer.class),
- isA(ReleaseState.class))).thenReturn(success());
+ Mockito.doNothing().when(schedulerService).setScheduleState(isA(User.class), isA(Long.class),
+ isA(Integer.class),
+ isA(ReleaseState.class));
MvcResult mvcResult = mockMvc.perform(post("/projects/{projectCode}/schedules/{id}/offline",123, 28)
.header(SESSION_ID, sessionId)
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java
index dd4d1e5ab1..c84057e9a2 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/SchedulerServiceTest.java
@@ -18,12 +18,14 @@
package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.impl.ProjectServiceImpl;
import org.apache.dolphinscheduler.api.service.impl.SchedulerServiceImpl;
-import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.model.Server;
+import org.apache.dolphinscheduler.dao.entity.Environment;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.User;
@@ -43,6 +45,8 @@ import java.util.Map;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
@@ -83,19 +87,31 @@ public class SchedulerServiceTest {
@Mock
private SchedulerApi schedulerApi;
- @Before
+ protected static User user;
+ protected Exception exception;
+ private static final String userName = "userName";
+ private static final String projectName = "projectName";
+ private static final long projectCode = 1L;
+ private static final int userId = 1;
+ private static final String processDefinitionName = "processDefinitionName";
+ private static final long processDefinitionCode = 2L;
+ private static final int processDefinitionVersion = 3;
+ private static final int scheduleId = 3;
+ private static final long environmentCode = 4L;
+ private static final String startTime = "2020-01-01 12:13:14";
+ private static final String endTime = "2020-02-01 12:13:14";
+ private static final String crontab = "0 0 * * * ? *";
+
+ @BeforeEach
public void setUp() {
-
+ user = new User();
+ user.setUserName(userName);
+ user.setId(userId);
}
@Test
public void testSetScheduleState() {
- String projectName = "test";
- long projectCode = 1L;
- User loginUser = new User();
- loginUser.setId(1);
- Map<String, Object> result = new HashMap<String, Object>();
- Project project = getProject(projectName, projectCode);
+ Project project = getProject();
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setProjectCode(projectCode);
@@ -109,53 +125,58 @@ public class SchedulerServiceTest {
masterServers.add(new Server());
Mockito.when(scheduleMapper.selectById(1)).thenReturn(schedule);
-
+ Mockito.when(processDefinitionMapper.queryByCode(1)).thenReturn(processDefinition);
Mockito.when(projectMapper.queryByCode(projectCode)).thenReturn(project);
Mockito.when(projectMapper.queryByName(projectName)).thenReturn(project);
- Mockito.when(processDefinitionMapper.queryByCode(1)).thenReturn(processDefinition);
-
- //hash no auth
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 1, ReleaseState.ONLINE);
-
- Mockito.when(projectService.hasProjectAndPerm(loginUser, project, result,null)).thenReturn(true);
- //schedule not exists
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 2, ReleaseState.ONLINE);
- Assert.assertEquals(Status.SCHEDULE_CRON_NOT_EXISTS, result.get(Constants.STATUS));
+ // schedule not exists
+ exception = Assertions.assertThrows(ServiceException.class, () -> {
+ schedulerService.setScheduleState(user, project.getCode(), 2, ReleaseState.ONLINE);
+ });
+ Assertions.assertEquals(Status.SCHEDULE_CRON_NOT_EXISTS.getCode(), ((ServiceException) exception).getCode());
- //SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 1, ReleaseState.OFFLINE);
- Assert.assertEquals(Status.SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE, result.get(Constants.STATUS));
+ // SCHEDULE_CRON_RELEASE_NEED_NOT_CHANGE
+ exception = Assertions.assertThrows(ServiceException.class, () -> {
+ schedulerService.setScheduleState(user, project.getCode(), 1, ReleaseState.OFFLINE);
+ });
+ Assertions.assertEquals(Status.SCHEDULE_CRON_REALEASE_NEED_NOT_CHANGE.getCode(),
+ ((ServiceException) exception).getCode());
//PROCESS_DEFINE_NOT_EXIST
schedule.setProcessDefinitionCode(2);
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 1, ReleaseState.ONLINE);
- Assert.assertEquals(Status.PROCESS_DEFINE_NOT_EXIST, result.get(Constants.STATUS));
+ exception = Assertions.assertThrows(ServiceException.class, () -> {
+ schedulerService.setScheduleState(user, project.getCode(), 1, ReleaseState.ONLINE);
+ });
+ Assertions.assertEquals(Status.PROCESS_DEFINE_NOT_EXIST.getCode(), ((ServiceException) exception).getCode());
schedule.setProcessDefinitionCode(1);
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 1, ReleaseState.ONLINE);
- Assert.assertEquals(Status.PROCESS_DAG_IS_EMPTY, result.get(Constants.STATUS));
-
+ // online also success
+ ProcessTaskRelation processTaskRelation = new ProcessTaskRelation();
+ List<ProcessTaskRelation> processTaskRelationList = new ArrayList<>();
+ processTaskRelationList.add(processTaskRelation);
+ Mockito.when(processTaskRelationMapper.queryByProcessCode(projectCode, 1)).thenReturn(processTaskRelationList);
+ exception = Assertions.assertThrows(ServiceException.class, () -> {
+ schedulerService.setScheduleState(user, project.getCode(), 1, ReleaseState.ONLINE);
+ });
+ Assertions.assertEquals(Status.PROCESS_DEFINE_NOT_RELEASE.getCode(), ((ServiceException) exception).getCode());
+
+ // SUCCESS
+ Server server = new Server();
+ List<Server> serverList = new ArrayList<>();
+ serverList.add(server);
+ Mockito.when(monitorService.getServerListFromRegistry(true)).thenReturn(serverList);
processDefinition.setReleaseState(ReleaseState.ONLINE);
- Mockito.when(processService.findProcessDefineById(1)).thenReturn(processDefinition);
-
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 1, ReleaseState.ONLINE);
- Assert.assertEquals(Status.PROCESS_DAG_IS_EMPTY, result.get(Constants.STATUS));
-
- //set master
- Mockito.when(monitorService.getServerListFromRegistry(true)).thenReturn(masterServers);
-
- //SUCCESS
- result = schedulerService.setScheduleState(loginUser, project.getCode(), 1, ReleaseState.ONLINE);
- Assert.assertEquals(Status.PROCESS_DAG_IS_EMPTY, result.get(Constants.STATUS));
+ Assertions.assertDoesNotThrow(() -> {
+ schedulerService.setScheduleState(user, project.getCode(), 1, ReleaseState.ONLINE);
+ });
}
- private Project getProject(String name, long code) {
+ private Project getProject() {
Project project = new Project();
- project.setName(name);
- project.setCode(code);
- project.setUserId(1);
-
+ project.setId(1);
+ project.setName(projectName);
+ project.setCode(projectCode);
+ project.setUserId(userId);
return project;
}
[dolphinscheduler] 01/02: cherry-pick Fix task group cannot release when kill task #13314
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kerwin pushed a commit to branch 3.1.3-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit 9564bdeefb06333f464b205f32c6b9522edb412b
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Tue Jan 3 09:52:03 2023 +0800
cherry-pick Fix task group cannot release when kill task #13314
---
.../master/runner/WorkflowExecuteRunnable.java | 45 ++++++++++++++--------
.../master/runner/task/CommonTaskProcessor.java | 10 +++--
.../service/process/ProcessServiceImpl.java | 8 +++-
.../plugin/task/api/utils/LogUtils.java | 45 +++++++++++++++++++++-
4 files changed, 86 insertions(+), 22 deletions(-)
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
index bfe6903213..939f4baca0 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
@@ -62,6 +62,7 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.DependResult;
import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.remote.command.HostUpdateCommand;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
@@ -341,23 +342,33 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
public boolean checkForceStartAndWakeUp(StateEvent stateEvent) {
TaskGroupQueue taskGroupQueue = this.processService.loadTaskGroupQueue(stateEvent.getTaskInstanceId());
if (taskGroupQueue.getForceStart() == Flag.YES.getCode()) {
+ logger.info("Begin to force start taskGroupQueue: {}", taskGroupQueue.getId());
TaskInstance taskInstance = this.processService.findTaskInstanceById(stateEvent.getTaskInstanceId());
ITaskProcessor taskProcessor = activeTaskProcessorMaps.get(taskInstance.getTaskCode());
taskProcessor.action(TaskAction.DISPATCH);
this.processService.updateTaskGroupQueueStatus(taskGroupQueue.getTaskId(),
TaskGroupQueueStatus.ACQUIRE_SUCCESS.getCode());
+ logger.info("Success force start taskGroupQueue: {}", taskGroupQueue.getId());
return true;
}
if (taskGroupQueue.getInQueue() == Flag.YES.getCode()) {
+ logger.info("Begin to wake up taskGroupQueue: {}", taskGroupQueue.getId());
boolean acquireTaskGroup = processService.robTaskGroupResource(taskGroupQueue);
if (acquireTaskGroup) {
TaskInstance taskInstance = this.processService.findTaskInstanceById(stateEvent.getTaskInstanceId());
ITaskProcessor taskProcessor = activeTaskProcessorMaps.get(taskInstance.getTaskCode());
taskProcessor.action(TaskAction.DISPATCH);
+ logger.info("Success wake up taskGroupQueue: {}", taskGroupQueue.getId());
return true;
}
+ logger.warn("Failed to wake up taskGroupQueue, taskGroupQueueId: {}", taskGroupQueue.getId());
+ return false;
+ } else {
+ logger.info(
+ "Failed to wake up the taskGroupQueue: {}, since the taskGroupQueue is not in queue, will no need to wake up.",
+ taskGroupQueue);
+ return true;
}
- return false;
}
public void processTimeout() {
@@ -425,7 +436,6 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
*
*/
public void releaseTaskGroup(TaskInstance taskInstance) {
- logger.info("Release task group");
if (taskInstance.getTaskGroupId() > 0) {
TaskInstance nextTaskInstance = this.processService.releaseTaskGroup(taskInstance);
if (nextTaskInstance != null) {
@@ -1765,19 +1775,24 @@ public class WorkflowExecuteRunnable implements Callable<WorkflowSubmitStatue> {
if (taskInstanceId == null || taskInstanceId.equals(0)) {
continue;
}
- TaskInstance taskInstance = processService.findTaskInstanceById(taskInstanceId);
- if (taskInstance == null || taskInstance.getState().isFinished()) {
- continue;
- }
- taskProcessor.action(TaskAction.STOP);
- if (taskProcessor.taskInstance().getState().isFinished()) {
- TaskStateEvent taskStateEvent = TaskStateEvent.builder()
- .processInstanceId(processInstance.getId())
- .taskInstanceId(taskInstance.getId())
- .status(taskProcessor.taskInstance().getState())
- .type(StateEventType.TASK_STATE_CHANGE)
- .build();
- this.addStateEvent(taskStateEvent);
+ LogUtils.setWorkflowAndTaskInstanceIDMDC(processInstance.getId(), taskInstanceId);
+ try {
+ TaskInstance taskInstance = processService.findTaskInstanceById(taskInstanceId);
+ if (taskInstance == null || taskInstance.getState().isFinished()) {
+ continue;
+ }
+ taskProcessor.action(TaskAction.STOP);
+ if (taskProcessor.taskInstance().getState().isFinished()) {
+ TaskStateEvent taskStateEvent = TaskStateEvent.builder()
+ .processInstanceId(processInstance.getId())
+ .taskInstanceId(taskInstance.getId())
+ .status(taskProcessor.taskInstance().getState())
+ .type(StateEventType.TASK_STATE_CHANGE)
+ .build();
+ this.addStateEvent(taskStateEvent);
+ }
+ } finally {
+ LogUtils.removeWorkflowAndTaskInstanceIdMDC();
}
}
}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java
index 73326720f6..5b2bf630e2 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java
@@ -138,11 +138,13 @@ public class CommonTaskProcessor extends BaseTaskProcessor {
public boolean killTask() {
try {
- taskInstance = processService.findTaskInstanceById(taskInstance.getId());
+ logger.info("Begin to kill task: {}", taskInstance.getName());
if (taskInstance == null) {
+ logger.warn("Kill task failed, the task instance is not exist");
return true;
}
if (taskInstance.getState().isFinished()) {
+ logger.warn("Kill task failed, the task instance is already finished");
return true;
}
// we don't wait the kill response
@@ -153,12 +155,12 @@ public class CommonTaskProcessor extends BaseTaskProcessor {
killRemoteTask();
}
} catch (Exception e) {
- logger.error("master kill task error, taskInstance id: {}", taskInstance.getId(), e);
+ logger.error("Master kill task: {} error, taskInstance id: {}", taskInstance.getName(),
+ taskInstance.getId(), e);
return false;
}
- logger.info("master success kill taskInstance name: {} taskInstance id: {}",
- taskInstance.getName(), taskInstance.getId());
+ logger.info("Master success kill task: {}, taskInstanceId: {}", taskInstance.getName(), taskInstance.getId());
return true;
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
index e580f1f74d..8e63b3128f 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
@@ -3040,6 +3040,10 @@ public class ProcessServiceImpl implements ProcessService {
logger.info("The taskGroupQueue's status is release, taskInstanceId: {}", taskInstance.getId());
return null;
}
+ if (thisTaskGroupQueue.getStatus() == TaskGroupQueueStatus.WAIT_QUEUE) {
+ logger.info("The taskGroupQueue's status is in waiting, will not need to release task group");
+ break;
+ }
} while (thisTaskGroupQueue.getForceStart() == Flag.NO.getCode()
&& taskGroupMapper.releaseTaskGroupResource(taskGroup.getId(),
taskGroup.getUseSize(),
@@ -3066,7 +3070,8 @@ public class ProcessServiceImpl implements ProcessService {
} while (this.taskGroupQueueMapper.updateInQueueCAS(Flag.NO.getCode(),
Flag.YES.getCode(),
taskGroupQueue.getId()) != 1);
- logger.info("Finished to release task group queue: taskGroupId: {}", taskInstance.getTaskGroupId());
+ logger.info("Finished to release task group queue: taskGroupId: {}, taskGroupQueueId: {}",
+ taskInstance.getTaskGroupId(), taskGroupQueue.getId());
return this.taskInstanceMapper.selectById(taskGroupQueue.getTaskId());
}
@@ -3080,6 +3085,7 @@ public class ProcessServiceImpl implements ProcessService {
@Override
public void changeTaskGroupQueueStatus(int taskId, TaskGroupQueueStatus status) {
TaskGroupQueue taskGroupQueue = taskGroupQueueMapper.queryByTaskId(taskId);
+ taskGroupQueue.setInQueue(Flag.NO.getCode());
taskGroupQueue.setStatus(status);
taskGroupQueue.setUpdateTime(new Date(System.currentTimeMillis()));
taskGroupQueueMapper.updateById(taskGroupQueue);
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java
index c3833c071a..c05c96dfcb 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java
@@ -17,10 +17,10 @@
package org.apache.dolphinscheduler.plugin.task.api.utils;
+import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
-import java.io.File;
-import java.io.IOException;
+import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
@@ -37,6 +37,7 @@ import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
+import org.slf4j.MDC;
@Slf4j
@UtilityClass
@@ -73,4 +74,44 @@ public class LogUtils {
return Collections.emptyList();
}
}
+
+ public static String readWholeFileContent(String filePath) {
+ String line;
+ StringBuilder sb = new StringBuilder();
+ try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)))) {
+ while ((line = br.readLine()) != null) {
+ sb.append(line + "\r\n");
+ }
+ return sb.toString();
+ } catch (IOException e) {
+ log.error("read file error", e);
+ }
+ return "";
+ }
+
+ public static void setWorkflowAndTaskInstanceIDMDC(Integer workflowInstanceId, Integer taskInstanceId) {
+ setWorkflowInstanceIdMDC(workflowInstanceId);
+ setTaskInstanceIdMDC(taskInstanceId);
+ }
+
+ public static void setWorkflowInstanceIdMDC(Integer workflowInstanceId) {
+ MDC.put(Constants.WORKFLOW_INSTANCE_ID_MDC_KEY, String.valueOf(workflowInstanceId));
+ }
+
+ public static void setTaskInstanceIdMDC(Integer taskInstanceId) {
+ MDC.put(Constants.TASK_INSTANCE_ID_MDC_KEY, String.valueOf(taskInstanceId));
+ }
+
+ public static void removeWorkflowAndTaskInstanceIdMDC() {
+ removeWorkflowInstanceIdMDC();
+ removeTaskInstanceIdMDC();
+ }
+
+ public static void removeWorkflowInstanceIdMDC() {
+ MDC.remove(Constants.WORKFLOW_INSTANCE_ID_MDC_KEY);
+ }
+
+ public static void removeTaskInstanceIdMDC() {
+ MDC.remove(Constants.TASK_INSTANCE_ID_MDC_KEY);
+ }
}